"Fossies" - the Fresh Open Source Software Archive

Member "keystone-17.0.0/keystone/tests/unit/test_limits.py" (13 May 2020, 73759 Bytes) of package /linux/misc/openstack/keystone-17.0.0.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 latest Fossies "Diffs" side-by-side code changes report for "test_limits.py": 16.0.1_vs_17.0.0.

    1 # Copyright 2018 Huawei
    2 #
    3 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    4 #    not use this file except in compliance with the License. You may obtain
    5 #    a copy of the License at
    6 #
    7 #         http://www.apache.org/licenses/LICENSE-2.0
    8 #
    9 #    Unless required by applicable law or agreed to in writing, software
   10 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   11 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   12 #    License for the specific language governing permissions and limitations
   13 #    under the License.
   14 
   15 import http.client
   16 import uuid
   17 
   18 from keystone.common import provider_api
   19 from keystone.common.validation import validators
   20 import keystone.conf
   21 from keystone.tests import unit
   22 from keystone.tests.unit import test_v3
   23 
   24 CONF = keystone.conf.CONF
   25 PROVIDERS = provider_api.ProviderAPIs
   26 
   27 
   28 class LimitModelTestCase(test_v3.RestfulTestCase):
   29 
   30     def test_get_default_limit_model_response_schema(self):
   31         schema = {
   32             'type': 'object',
   33             'properties': {
   34                 'model': {
   35                     'type': 'object',
   36                     'properties': {
   37                         'name': {'type': 'string'},
   38                         'description': {'type': 'string'}
   39                     },
   40                     'required': ['name', 'description'],
   41                     'additionalProperties': False,
   42                 },
   43             },
   44             'required': ['model'],
   45             'additionalProperties': False,
   46         }
   47         validator = validators.SchemaValidator(schema)
   48         response = self.get('/limits/model')
   49         validator.validate(response.json_body)
   50 
   51     def test_head_limit_model(self):
   52         self.head('/limits/model', expected_status=http.client.OK)
   53 
   54     def test_get_limit_model_returns_default_model(self):
   55         response = self.get('/limits/model')
   56         model = response.result
   57         expected = {
   58             'model': {
   59                 'name': 'flat',
   60                 'description': (
   61                     'Limit enforcement and validation does not take project '
   62                     'hierarchy into consideration.'
   63                 )
   64             }
   65         }
   66         self.assertDictEqual(expected, model)
   67 
   68     def test_get_limit_model_without_token_fails(self):
   69         self.get(
   70             '/limits/model', noauth=True,
   71             expected_status=http.client.UNAUTHORIZED
   72         )
   73 
   74     def test_head_limit_model_without_token_fails(self):
   75         self.head(
   76             '/limits/model', noauth=True,
   77             expected_status=http.client.UNAUTHORIZED
   78         )
   79 
   80 
   81 class RegisteredLimitsTestCase(test_v3.RestfulTestCase):
   82     """Test registered_limits CRUD."""
   83 
   84     def setUp(self):
   85         super(RegisteredLimitsTestCase, self).setUp()
   86 
   87         # Most of these tests require system-scoped tokens. Let's have one on
   88         # hand so that we can use it in tests when we need it.
   89         PROVIDERS.assignment_api.create_system_grant_for_user(
   90             self.user_id, self.role_id
   91         )
   92         self.system_admin_token = self.get_system_scoped_token()
   93 
   94         # There is already a sample service and region created from
   95         # load_sample_data() but we're going to create another service and
   96         # region for specific testing purposes.
   97         response = self.post('/regions', body={'region': {}})
   98         self.region2 = response.json_body['region']
   99         self.region_id2 = self.region2['id']
  100 
  101         service_ref = {'service': {
  102             'name': uuid.uuid4().hex,
  103             'enabled': True,
  104             'type': 'type2'}}
  105         response = self.post('/services', body=service_ref)
  106         self.service2 = response.json_body['service']
  107         self.service_id2 = self.service2['id']
  108 
  109     def test_create_registered_limit(self):
  110         ref = unit.new_registered_limit_ref(service_id=self.service_id,
  111                                             region_id=self.region_id)
  112         r = self.post(
  113             '/registered_limits',
  114             body={'registered_limits': [ref]},
  115             token=self.system_admin_token,
  116             expected_status=http.client.CREATED)
  117         registered_limits = r.result['registered_limits']
  118         for key in ['service_id', 'region_id', 'resource_name',
  119                     'default_limit', 'description']:
  120             self.assertEqual(registered_limits[0][key], ref[key])
  121 
  122     def test_create_registered_limit_without_region(self):
  123         ref = unit.new_registered_limit_ref(service_id=self.service_id)
  124         r = self.post(
  125             '/registered_limits',
  126             body={'registered_limits': [ref]},
  127             token=self.system_admin_token,
  128             expected_status=http.client.CREATED)
  129         registered_limits = r.result['registered_limits']
  130         for key in ['service_id', 'resource_name', 'default_limit']:
  131             self.assertEqual(registered_limits[0][key], ref[key])
  132         self.assertIsNone(registered_limits[0].get('region_id'))
  133 
  134     def test_create_registered_without_description(self):
  135         ref = unit.new_registered_limit_ref(service_id=self.service_id,
  136                                             region_id=self.region_id)
  137         ref.pop('description')
  138         r = self.post(
  139             '/registered_limits',
  140             body={'registered_limits': [ref]},
  141             token=self.system_admin_token,
  142             expected_status=http.client.CREATED)
  143         registered_limits = r.result['registered_limits']
  144         for key in ['service_id', 'region_id', 'resource_name',
  145                     'default_limit']:
  146             self.assertEqual(registered_limits[0][key], ref[key])
  147         self.assertIsNone(registered_limits[0]['description'])
  148 
  149     def test_create_multi_registered_limit(self):
  150         ref1 = unit.new_registered_limit_ref(service_id=self.service_id,
  151                                              region_id=self.region_id,
  152                                              resource_name='volume')
  153         ref2 = unit.new_registered_limit_ref(service_id=self.service_id,
  154                                              resource_name='snapshot')
  155         r = self.post(
  156             '/registered_limits',
  157             body={'registered_limits': [ref1, ref2]},
  158             token=self.system_admin_token,
  159             expected_status=http.client.CREATED)
  160         registered_limits = r.result['registered_limits']
  161         for key in ['service_id', 'resource_name', 'default_limit']:
  162             self.assertEqual(registered_limits[0][key], ref1[key])
  163             self.assertEqual(registered_limits[1][key], ref2[key])
  164         self.assertEqual(registered_limits[0]['region_id'], ref1['region_id'])
  165         self.assertIsNone(registered_limits[1].get('region_id'))
  166 
  167     def test_create_registered_limit_return_count(self):
  168         ref1 = unit.new_registered_limit_ref(service_id=self.service_id,
  169                                              region_id=self.region_id)
  170         r = self.post(
  171             '/registered_limits',
  172             body={'registered_limits': [ref1]},
  173             token=self.system_admin_token,
  174             expected_status=http.client.CREATED)
  175         registered_limits = r.result['registered_limits']
  176         self.assertEqual(1, len(registered_limits))
  177 
  178         ref2 = unit.new_registered_limit_ref(service_id=self.service_id2,
  179                                              region_id=self.region_id2)
  180         ref3 = unit.new_registered_limit_ref(service_id=self.service_id2)
  181         r = self.post(
  182             '/registered_limits',
  183             body={'registered_limits': [ref2, ref3]},
  184             token=self.system_admin_token,
  185             expected_status=http.client.CREATED)
  186         registered_limits = r.result['registered_limits']
  187         self.assertEqual(2, len(registered_limits))
  188 
  189     def test_create_registered_limit_with_invalid_input(self):
  190         ref1 = unit.new_registered_limit_ref()
  191         ref2 = unit.new_registered_limit_ref(default_limit='not_int')
  192         ref3 = unit.new_registered_limit_ref(resource_name=123)
  193         ref4 = unit.new_registered_limit_ref(region_id='fake_region')
  194         for input_limit in [ref1, ref2, ref3, ref4]:
  195             self.post(
  196                 '/registered_limits',
  197                 body={'registered_limits': [input_limit]},
  198                 token=self.system_admin_token,
  199                 expected_status=http.client.BAD_REQUEST)
  200 
  201     def test_create_registered_limit_duplicate(self):
  202         ref = unit.new_registered_limit_ref(service_id=self.service_id,
  203                                             region_id=self.region_id)
  204         self.post(
  205             '/registered_limits',
  206             body={'registered_limits': [ref]},
  207             token=self.system_admin_token,
  208             expected_status=http.client.CREATED)
  209         self.post(
  210             '/registered_limits',
  211             body={'registered_limits': [ref]},
  212             token=self.system_admin_token,
  213             expected_status=http.client.CONFLICT)
  214 
  215     def test_update_registered_limit(self):
  216         ref = unit.new_registered_limit_ref(service_id=self.service_id,
  217                                             region_id=self.region_id,
  218                                             resource_name='volume',
  219                                             default_limit=10)
  220         r = self.post(
  221             '/registered_limits',
  222             body={'registered_limits': [ref]},
  223             token=self.system_admin_token,
  224             expected_status=http.client.CREATED)
  225         update_ref = {
  226             'service_id': self.service_id2,
  227             'region_id': self.region_id2,
  228             'resource_name': 'snapshot',
  229             'default_limit': 5,
  230             'description': 'test description'
  231         }
  232         r = self.patch(
  233             '/registered_limits/%s' % r.result['registered_limits'][0]['id'],
  234             body={'registered_limit': update_ref},
  235             token=self.system_admin_token,
  236             expected_status=http.client.OK)
  237         new_registered_limits = r.result['registered_limit']
  238 
  239         self.assertEqual(new_registered_limits['service_id'], self.service_id2)
  240         self.assertEqual(new_registered_limits['region_id'], self.region_id2)
  241         self.assertEqual(new_registered_limits['resource_name'], 'snapshot')
  242         self.assertEqual(new_registered_limits['default_limit'], 5)
  243         self.assertEqual(new_registered_limits['description'],
  244                          'test description')
  245 
  246     def test_update_registered_limit_region_failed(self):
  247         ref = unit.new_registered_limit_ref(service_id=self.service_id,
  248                                             resource_name='volume',
  249                                             default_limit=10,
  250                                             description='test description')
  251         r = self.post(
  252             '/registered_limits',
  253             body={'registered_limits': [ref]},
  254             token=self.system_admin_token,
  255             expected_status=http.client.CREATED)
  256         update_ref = {
  257             'region_id': self.region_id,
  258         }
  259         registered_limit_id = r.result['registered_limits'][0]['id']
  260         r = self.patch(
  261             '/registered_limits/%s' % registered_limit_id,
  262             body={'registered_limit': update_ref},
  263             token=self.system_admin_token,
  264             expected_status=http.client.OK)
  265         new_registered_limits = r.result['registered_limit']
  266         self.assertEqual(self.region_id, new_registered_limits['region_id'])
  267 
  268         update_ref['region_id'] = ''
  269         r = self.patch(
  270             '/registered_limits/%s' % registered_limit_id,
  271             body={'registered_limit': update_ref},
  272             token=self.system_admin_token,
  273             expected_status=http.client.BAD_REQUEST)
  274 
  275     def test_update_registered_limit_description(self):
  276         ref = unit.new_registered_limit_ref(service_id=self.service_id,
  277                                             region_id=self.region_id,
  278                                             resource_name='volume',
  279                                             default_limit=10)
  280         r = self.post(
  281             '/registered_limits',
  282             body={'registered_limits': [ref]},
  283             token=self.system_admin_token,
  284             expected_status=http.client.CREATED)
  285         update_ref = {
  286             'description': 'test description'
  287         }
  288         registered_limit_id = r.result['registered_limits'][0]['id']
  289         r = self.patch(
  290             '/registered_limits/%s' % registered_limit_id,
  291             body={'registered_limit': update_ref},
  292             token=self.system_admin_token,
  293             expected_status=http.client.OK)
  294         new_registered_limits = r.result['registered_limit']
  295         self.assertEqual(new_registered_limits['description'],
  296                          'test description')
  297 
  298         update_ref['description'] = ''
  299         r = self.patch(
  300             '/registered_limits/%s' % registered_limit_id,
  301             body={'registered_limit': update_ref},
  302             token=self.system_admin_token,
  303             expected_status=http.client.OK)
  304         new_registered_limits = r.result['registered_limit']
  305         self.assertEqual(new_registered_limits['description'], '')
  306 
  307     def test_update_registered_limit_region_id_to_none(self):
  308         ref = unit.new_registered_limit_ref(service_id=self.service_id,
  309                                             region_id=self.region_id,
  310                                             resource_name='volume',
  311                                             default_limit=10)
  312         r = self.post(
  313             '/registered_limits',
  314             body={'registered_limits': [ref]},
  315             token=self.system_admin_token,
  316             expected_status=http.client.CREATED)
  317         update_ref = {
  318             'region_id': None
  319         }
  320         registered_limit_id = r.result['registered_limits'][0]['id']
  321         r = self.patch(
  322             '/registered_limits/%s' % registered_limit_id,
  323             body={'registered_limit': update_ref},
  324             token=self.system_admin_token,
  325             expected_status=http.client.OK)
  326         self.assertIsNone(r.result['registered_limit']['region_id'])
  327 
  328     def test_update_registered_limit_region_id_to_none_conflict(self):
  329         ref1 = unit.new_registered_limit_ref(service_id=self.service_id,
  330                                              resource_name='volume',
  331                                              default_limit=10)
  332         ref2 = unit.new_registered_limit_ref(service_id=self.service_id,
  333                                              region_id=self.region_id,
  334                                              resource_name='volume',
  335                                              default_limit=10)
  336         self.post(
  337             '/registered_limits',
  338             body={'registered_limits': [ref1]},
  339             token=self.system_admin_token,
  340             expected_status=http.client.CREATED)
  341         r = self.post(
  342             '/registered_limits',
  343             body={'registered_limits': [ref2]},
  344             token=self.system_admin_token,
  345             expected_status=http.client.CREATED)
  346 
  347         update_ref = {
  348             'region_id': None
  349         }
  350         registered_limit_id = r.result['registered_limits'][0]['id']
  351         # There is a registered limit with "service_id=self.service_id,
  352         # region_id=None" already. So update ref2's region_id to None will
  353         # raise 409 Conflict Error.
  354         self.patch(
  355             '/registered_limits/%s' % registered_limit_id,
  356             body={'registered_limit': update_ref},
  357             token=self.system_admin_token,
  358             expected_status=http.client.CONFLICT)
  359 
  360     def test_update_registered_limit_not_found(self):
  361         update_ref = {
  362             'service_id': self.service_id,
  363             'region_id': self.region_id,
  364             'resource_name': 'snapshot',
  365             'default_limit': 5
  366         }
  367         self.patch(
  368             '/registered_limits/%s' % uuid.uuid4().hex,
  369             body={'registered_limit': update_ref},
  370             token=self.system_admin_token,
  371             expected_status=http.client.NOT_FOUND)
  372 
  373     def test_update_registered_limit_with_invalid_input(self):
  374         ref = unit.new_registered_limit_ref(service_id=self.service_id,
  375                                             region_id=self.region_id,
  376                                             resource_name='volume',
  377                                             default_limit=10)
  378         r = self.post(
  379             '/registered_limits',
  380             body={'registered_limits': [ref]},
  381             token=self.system_admin_token,
  382             expected_status=http.client.CREATED)
  383         reg_id = r.result['registered_limits'][0]['id']
  384 
  385         update_ref1 = unit.new_registered_limit_ref(service_id='fake_id')
  386         update_ref2 = unit.new_registered_limit_ref(default_limit='not_int')
  387         update_ref3 = unit.new_registered_limit_ref(resource_name=123)
  388         update_ref4 = unit.new_registered_limit_ref(region_id='fake_region')
  389         update_ref5 = unit.new_registered_limit_ref(description=123)
  390         for input_limit in [update_ref1, update_ref2, update_ref3,
  391                             update_ref4, update_ref5]:
  392             self.patch(
  393                 '/registered_limits/%s' % reg_id,
  394                 body={'registered_limit': input_limit},
  395                 token=self.system_admin_token,
  396                 expected_status=http.client.BAD_REQUEST)
  397 
  398     def test_update_registered_limit_with_referenced_limit(self):
  399         ref = unit.new_registered_limit_ref(service_id=self.service_id,
  400                                             region_id=self.region_id,
  401                                             resource_name='volume',
  402                                             default_limit=10)
  403         r = self.post(
  404             '/registered_limits',
  405             body={'registered_limits': [ref]},
  406             token=self.system_admin_token,
  407             expected_status=http.client.CREATED)
  408 
  409         ref = unit.new_limit_ref(project_id=self.project_id,
  410                                  service_id=self.service_id,
  411                                  region_id=self.region_id,
  412                                  resource_name='volume')
  413         self.post(
  414             '/limits',
  415             body={'limits': [ref]},
  416             token=self.system_admin_token,
  417             expected_status=http.client.CREATED)
  418 
  419         update_ref = {
  420             'service_id': self.service_id2,
  421             'region_id': self.region_id2,
  422             'resource_name': 'snapshot',
  423             'default_limit': 5
  424         }
  425         self.patch(
  426             '/registered_limits/%s' % r.result['registered_limits'][0]['id'],
  427             body={'registered_limit': update_ref},
  428             token=self.system_admin_token,
  429             expected_status=http.client.FORBIDDEN)
  430 
  431     def test_list_registered_limit(self):
  432         r = self.get(
  433             '/registered_limits',
  434             expected_status=http.client.OK)
  435         self.assertEqual([], r.result.get('registered_limits'))
  436 
  437         ref1 = unit.new_registered_limit_ref(service_id=self.service_id,
  438                                              resource_name='test_resource',
  439                                              region_id=self.region_id)
  440         ref2 = unit.new_registered_limit_ref(service_id=self.service_id2,
  441                                              resource_name='test_resource',
  442                                              region_id=self.region_id2)
  443         r = self.post(
  444             '/registered_limits',
  445             body={'registered_limits': [ref1, ref2]},
  446             token=self.system_admin_token,
  447             expected_status=http.client.CREATED)
  448         id1 = r.result['registered_limits'][0]['id']
  449         r = self.get(
  450             '/registered_limits',
  451             expected_status=http.client.OK)
  452         registered_limits = r.result['registered_limits']
  453         self.assertEqual(len(registered_limits), 2)
  454         for key in ['service_id', 'region_id', 'resource_name',
  455                     'default_limit']:
  456             if registered_limits[0]['id'] == id1:
  457                 self.assertEqual(registered_limits[0][key], ref1[key])
  458                 self.assertEqual(registered_limits[1][key], ref2[key])
  459                 break
  460             self.assertEqual(registered_limits[1][key], ref1[key])
  461             self.assertEqual(registered_limits[0][key], ref2[key])
  462 
  463         r = self.get(
  464             '/registered_limits?service_id=%s' % self.service_id,
  465             expected_status=http.client.OK)
  466         registered_limits = r.result['registered_limits']
  467         self.assertEqual(len(registered_limits), 1)
  468         for key in ['service_id', 'region_id', 'resource_name',
  469                     'default_limit']:
  470             self.assertEqual(registered_limits[0][key], ref1[key])
  471 
  472         r = self.get(
  473             '/registered_limits?region_id=%s' % self.region_id2,
  474             expected_status=http.client.OK)
  475         registered_limits = r.result['registered_limits']
  476         self.assertEqual(len(registered_limits), 1)
  477         for key in ['service_id', 'region_id', 'resource_name',
  478                     'default_limit']:
  479             self.assertEqual(registered_limits[0][key], ref2[key])
  480 
  481         r = self.get(
  482             '/registered_limits?resource_name=test_resource',
  483             expected_status=http.client.OK)
  484         registered_limits = r.result['registered_limits']
  485         self.assertEqual(len(registered_limits), 2)
  486 
  487     def test_show_registered_limit(self):
  488         ref1 = unit.new_registered_limit_ref(service_id=self.service_id,
  489                                              region_id=self.region_id)
  490         ref2 = unit.new_registered_limit_ref(service_id=self.service_id2,
  491                                              region_id=self.region_id2)
  492         r = self.post(
  493             '/registered_limits',
  494             body={'registered_limits': [ref1, ref2]},
  495             token=self.system_admin_token,
  496             expected_status=http.client.CREATED)
  497         id1 = r.result['registered_limits'][0]['id']
  498         self.get(
  499             '/registered_limits/fake_id',
  500             expected_status=http.client.NOT_FOUND)
  501         r = self.get(
  502             '/registered_limits/%s' % id1,
  503             expected_status=http.client.OK)
  504         registered_limit = r.result['registered_limit']
  505         for key in ['service_id', 'region_id', 'resource_name',
  506                     'default_limit', 'description']:
  507             self.assertEqual(registered_limit[key], ref1[key])
  508 
  509     def test_delete_registered_limit(self):
  510         ref1 = unit.new_registered_limit_ref(service_id=self.service_id,
  511                                              region_id=self.region_id)
  512         ref2 = unit.new_registered_limit_ref(service_id=self.service_id2,
  513                                              region_id=self.region_id2)
  514         r = self.post(
  515             '/registered_limits',
  516             body={'registered_limits': [ref1, ref2]},
  517             token=self.system_admin_token,
  518             expected_status=http.client.CREATED)
  519         id1 = r.result['registered_limits'][0]['id']
  520         self.delete('/registered_limits/%s' % id1,
  521                     token=self.system_admin_token,
  522                     expected_status=http.client.NO_CONTENT)
  523         self.delete('/registered_limits/fake_id',
  524                     token=self.system_admin_token,
  525                     expected_status=http.client.NOT_FOUND)
  526         r = self.get(
  527             '/registered_limits',
  528             expected_status=http.client.OK)
  529         registered_limits = r.result['registered_limits']
  530         self.assertEqual(len(registered_limits), 1)
  531 
  532     def test_delete_registered_limit_with_referenced_limit(self):
  533         ref = unit.new_registered_limit_ref(service_id=self.service_id,
  534                                             region_id=self.region_id,
  535                                             resource_name='volume',
  536                                             default_limit=10)
  537         r = self.post(
  538             '/registered_limits',
  539             body={'registered_limits': [ref]},
  540             token=self.system_admin_token,
  541             expected_status=http.client.CREATED)
  542 
  543         ref = unit.new_limit_ref(project_id=self.project_id,
  544                                  service_id=self.service_id,
  545                                  region_id=self.region_id,
  546                                  resource_name='volume')
  547         self.post(
  548             '/limits',
  549             body={'limits': [ref]},
  550             token=self.system_admin_token,
  551             expected_status=http.client.CREATED)
  552 
  553         id = r.result['registered_limits'][0]['id']
  554         self.delete('/registered_limits/%s' % id,
  555                     expected_status=http.client.FORBIDDEN)
  556 
  557 
  558 class LimitsTestCase(test_v3.RestfulTestCase):
  559     """Test limits CRUD."""
  560 
  561     def setUp(self):
  562         super(LimitsTestCase, self).setUp()
  563         # FIXME(lbragstad): Remove all this duplicated logic once we get all
  564         # keystone tests using bootstrap consistently. This is something the
  565         # bootstrap utility already does for us.
  566         reader_role = {'id': uuid.uuid4().hex, 'name': 'reader'}
  567         reader_role = PROVIDERS.role_api.create_role(
  568             reader_role['id'], reader_role
  569         )
  570 
  571         member_role = {'id': uuid.uuid4().hex, 'name': 'member'}
  572         member_role = PROVIDERS.role_api.create_role(
  573             member_role['id'], member_role
  574         )
  575         PROVIDERS.role_api.create_implied_role(self.role_id, member_role['id'])
  576         PROVIDERS.role_api.create_implied_role(
  577             member_role['id'], reader_role['id']
  578         )
  579 
  580         # Most of these tests require system-scoped tokens. Let's have one on
  581         # hand so that we can use it in tests when we need it.
  582         PROVIDERS.assignment_api.create_system_grant_for_user(
  583             self.user_id, self.role_id
  584         )
  585         self.system_admin_token = self.get_system_scoped_token()
  586 
  587         # There is already a sample service and region created from
  588         # load_sample_data() but we're going to create another service and
  589         # region for specific testing purposes.
  590         response = self.post('/regions', body={'region': {}})
  591         self.region2 = response.json_body['region']
  592         self.region_id2 = self.region2['id']
  593 
  594         service_ref = {'service': {
  595             'name': uuid.uuid4().hex,
  596             'enabled': True,
  597             'type': 'type2'}}
  598         response = self.post('/services', body=service_ref)
  599         self.service2 = response.json_body['service']
  600         self.service_id2 = self.service2['id']
  601 
  602         ref1 = unit.new_registered_limit_ref(service_id=self.service_id,
  603                                              region_id=self.region_id,
  604                                              resource_name='volume')
  605         ref2 = unit.new_registered_limit_ref(service_id=self.service_id2,
  606                                              resource_name='snapshot')
  607         ref3 = unit.new_registered_limit_ref(service_id=self.service_id,
  608                                              region_id=self.region_id,
  609                                              resource_name='backup')
  610         self.post(
  611             '/registered_limits',
  612             body={'registered_limits': [ref1, ref2, ref3]},
  613             token=self.system_admin_token,
  614             expected_status=http.client.CREATED)
  615 
  616         # Create more assignments, all are:
  617         #
  618         # self.user -- admin -- self.project
  619         # self.user -- non-admin -- self.project_2
  620         # self.user -- admin -- self.domain
  621         # self.user -- non-admin -- self.domain_2
  622         # self.user -- admin -- system
  623         self.project_2 = unit.new_project_ref(domain_id=self.domain_id)
  624         self.project_2_id = self.project_2['id']
  625         PROVIDERS.resource_api.create_project(self.project_2_id,
  626                                               self.project_2)
  627 
  628         self.domain_2 = unit.new_domain_ref()
  629         self.domain_2_id = self.domain_2['id']
  630         PROVIDERS.resource_api.create_domain(self.domain_2_id, self.domain_2)
  631 
  632         self.role_2 = unit.new_role_ref(name='non-admin')
  633         self.role_2_id = self.role_2['id']
  634         PROVIDERS.role_api.create_role(self.role_2_id, self.role_2)
  635 
  636         PROVIDERS.assignment_api.create_grant(
  637             self.role_2_id, user_id=self.user_id, project_id=self.project_2_id)
  638         PROVIDERS.assignment_api.create_grant(
  639             self.role_id, user_id=self.user_id, domain_id=self.domain_id)
  640         PROVIDERS.assignment_api.create_grant(
  641             self.role_2_id, user_id=self.user_id, domain_id=self.domain_2_id)
  642         PROVIDERS.assignment_api.create_system_grant_for_user(
  643             self.user_id, self.role_id)
  644 
  645     def test_create_project_limit(self):
  646         ref = unit.new_limit_ref(project_id=self.project_id,
  647                                  service_id=self.service_id,
  648                                  region_id=self.region_id,
  649                                  resource_name='volume')
  650         r = self.post(
  651             '/limits',
  652             body={'limits': [ref]},
  653             token=self.system_admin_token,
  654             expected_status=http.client.CREATED)
  655         limits = r.result['limits']
  656 
  657         self.assertIsNotNone(limits[0]['id'])
  658         self.assertIsNone(limits[0]['domain_id'])
  659         for key in ['service_id', 'region_id', 'resource_name',
  660                     'resource_limit', 'description', 'project_id']:
  661             self.assertEqual(limits[0][key], ref[key])
  662 
  663     def test_create_domain_limit(self):
  664         ref = unit.new_limit_ref(domain_id=self.domain_id,
  665                                  service_id=self.service_id,
  666                                  region_id=self.region_id,
  667                                  resource_name='volume')
  668         r = self.post(
  669             '/limits',
  670             body={'limits': [ref]},
  671             token=self.system_admin_token,
  672             expected_status=http.client.CREATED)
  673         limits = r.result['limits']
  674 
  675         self.assertIsNotNone(limits[0]['id'])
  676         self.assertIsNone(limits[0]['project_id'])
  677         for key in ['service_id', 'region_id', 'resource_name',
  678                     'resource_limit', 'description', 'domain_id']:
  679             self.assertEqual(limits[0][key], ref[key])
  680 
  681     def test_create_limit_without_region(self):
  682         ref = unit.new_limit_ref(project_id=self.project_id,
  683                                  service_id=self.service_id2,
  684                                  resource_name='snapshot')
  685         r = self.post(
  686             '/limits',
  687             body={'limits': [ref]},
  688             token=self.system_admin_token,
  689             expected_status=http.client.CREATED)
  690         limits = r.result['limits']
  691 
  692         self.assertIsNotNone(limits[0]['id'])
  693         self.assertIsNotNone(limits[0]['project_id'])
  694         for key in ['service_id', 'resource_name', 'resource_limit']:
  695             self.assertEqual(limits[0][key], ref[key])
  696         self.assertIsNone(limits[0].get('region_id'))
  697 
  698     def test_create_limit_without_description(self):
  699         ref = unit.new_limit_ref(project_id=self.project_id,
  700                                  service_id=self.service_id,
  701                                  region_id=self.region_id,
  702                                  resource_name='volume')
  703         ref.pop('description')
  704         r = self.post(
  705             '/limits',
  706             body={'limits': [ref]},
  707             token=self.system_admin_token,
  708             expected_status=http.client.CREATED)
  709         limits = r.result['limits']
  710 
  711         self.assertIsNotNone(limits[0]['id'])
  712         self.assertIsNotNone(limits[0]['project_id'])
  713         for key in ['service_id', 'region_id', 'resource_name',
  714                     'resource_limit']:
  715             self.assertEqual(limits[0][key], ref[key])
  716         self.assertIsNone(limits[0]['description'])
  717 
  718     def test_create_limit_with_domain_as_project(self):
  719         ref = unit.new_limit_ref(project_id=self.domain_id,
  720                                  service_id=self.service_id,
  721                                  region_id=self.region_id,
  722                                  resource_name='volume')
  723         r = self.post('/limits', body={'limits': [ref]},
  724                       token=self.system_admin_token)
  725         limits = r.result['limits']
  726         self.assertIsNone(limits[0]['project_id'])
  727         self.assertEqual(self.domain_id, limits[0]['domain_id'])
  728 
  729     def test_create_multi_limit(self):
  730         ref1 = unit.new_limit_ref(project_id=self.project_id,
  731                                   service_id=self.service_id,
  732                                   region_id=self.region_id,
  733                                   resource_name='volume')
  734         ref2 = unit.new_limit_ref(project_id=self.project_id,
  735                                   service_id=self.service_id2,
  736                                   resource_name='snapshot')
  737         r = self.post(
  738             '/limits',
  739             body={'limits': [ref1, ref2]},
  740             token=self.system_admin_token,
  741             expected_status=http.client.CREATED)
  742         limits = r.result['limits']
  743         for key in ['service_id', 'resource_name', 'resource_limit']:
  744             self.assertEqual(limits[0][key], ref1[key])
  745             self.assertEqual(limits[1][key], ref2[key])
  746         self.assertEqual(limits[0]['region_id'], ref1['region_id'])
  747         self.assertIsNone(limits[1].get('region_id'))
  748 
  749     def test_create_limit_return_count(self):
  750         ref1 = unit.new_limit_ref(project_id=self.project_id,
  751                                   service_id=self.service_id,
  752                                   region_id=self.region_id,
  753                                   resource_name='volume')
  754         r = self.post(
  755             '/limits',
  756             body={'limits': [ref1]},
  757             token=self.system_admin_token,
  758             expected_status=http.client.CREATED)
  759         limits = r.result['limits']
  760         self.assertEqual(1, len(limits))
  761 
  762         ref2 = unit.new_limit_ref(project_id=self.project_id,
  763                                   service_id=self.service_id2,
  764                                   resource_name='snapshot')
  765         ref3 = unit.new_limit_ref(project_id=self.project_id,
  766                                   service_id=self.service_id,
  767                                   region_id=self.region_id,
  768                                   resource_name='backup')
  769         r = self.post(
  770             '/limits',
  771             body={'limits': [ref2, ref3]},
  772             token=self.system_admin_token,
  773             expected_status=http.client.CREATED)
  774         limits = r.result['limits']
  775         self.assertEqual(2, len(limits))
  776 
  777     def test_create_limit_with_invalid_input(self):
  778         ref1 = unit.new_limit_ref(project_id=self.project_id,
  779                                   resource_limit='not_int')
  780         ref2 = unit.new_limit_ref(project_id=self.project_id,
  781                                   resource_name=123)
  782         ref3 = unit.new_limit_ref(project_id=self.project_id,
  783                                   region_id='fake_region')
  784         for input_limit in [ref1, ref2, ref3]:
  785             self.post(
  786                 '/limits',
  787                 body={'limits': [input_limit]},
  788                 token=self.system_admin_token,
  789                 expected_status=http.client.BAD_REQUEST)
  790 
  791     def test_create_limit_duplicate(self):
  792         ref = unit.new_limit_ref(project_id=self.project_id,
  793                                  service_id=self.service_id,
  794                                  region_id=self.region_id,
  795                                  resource_name='volume')
  796         self.post(
  797             '/limits',
  798             body={'limits': [ref]},
  799             token=self.system_admin_token,
  800             expected_status=http.client.CREATED)
  801         self.post(
  802             '/limits',
  803             body={'limits': [ref]},
  804             token=self.system_admin_token,
  805             expected_status=http.client.CONFLICT)
  806 
  807     def test_create_limit_without_reference_registered_limit(self):
  808         ref = unit.new_limit_ref(project_id=self.project_id,
  809                                  service_id=self.service_id,
  810                                  region_id=self.region_id2,
  811                                  resource_name='volume')
  812         self.post(
  813             '/limits',
  814             body={'limits': [ref]},
  815             token=self.system_admin_token,
  816             expected_status=http.client.FORBIDDEN)
  817 
  818     def test_update_limit(self):
  819         ref = unit.new_limit_ref(project_id=self.project_id,
  820                                  service_id=self.service_id,
  821                                  region_id=self.region_id,
  822                                  resource_name='volume',
  823                                  resource_limit=10)
  824         r = self.post(
  825             '/limits',
  826             body={'limits': [ref]},
  827             token=self.system_admin_token,
  828             expected_status=http.client.CREATED)
  829         update_ref = {
  830             'resource_limit': 5,
  831             'description': 'test description'
  832         }
  833         r = self.patch(
  834             '/limits/%s' % r.result['limits'][0]['id'],
  835             body={'limit': update_ref},
  836             token=self.system_admin_token,
  837             expected_status=http.client.OK)
  838         new_limits = r.result['limit']
  839 
  840         self.assertEqual(new_limits['resource_limit'], 5)
  841         self.assertEqual(new_limits['description'], 'test description')
  842 
  843     def test_update_limit_not_found(self):
  844         update_ref = {
  845             'resource_limit': 5
  846         }
  847         self.patch(
  848             '/limits/%s' % uuid.uuid4().hex,
  849             body={'limit': update_ref},
  850             token=self.system_admin_token,
  851             expected_status=http.client.NOT_FOUND)
  852 
  853     def test_update_limit_with_invalid_input(self):
  854         ref = unit.new_limit_ref(project_id=self.project_id,
  855                                  service_id=self.service_id,
  856                                  region_id=self.region_id,
  857                                  resource_name='volume',
  858                                  resource_limit=10)
  859         r = self.post(
  860             '/limits',
  861             body={'limits': [ref]},
  862             token=self.system_admin_token,
  863             expected_status=http.client.CREATED)
  864         limit_id = r.result['limits'][0]['id']
  865 
  866         invalid_resource_limit_update = {
  867             'resource_limit': 'not_int'
  868         }
  869         invalid_description_update = {
  870             'description': 123
  871         }
  872         for input_limit in [invalid_resource_limit_update,
  873                             invalid_description_update]:
  874             self.patch(
  875                 '/limits/%s' % limit_id,
  876                 body={'limit': input_limit},
  877                 token=self.system_admin_token,
  878                 expected_status=http.client.BAD_REQUEST)
  879 
  880     def test_list_limit(self):
  881         r = self.get(
  882             '/limits',
  883             token=self.system_admin_token,
  884             expected_status=http.client.OK)
  885         self.assertEqual([], r.result.get('limits'))
  886 
  887         ref1 = unit.new_limit_ref(project_id=self.project_id,
  888                                   service_id=self.service_id,
  889                                   region_id=self.region_id,
  890                                   resource_name='volume')
  891         ref2 = unit.new_limit_ref(project_id=self.project_id,
  892                                   service_id=self.service_id2,
  893                                   resource_name='snapshot')
  894         r = self.post(
  895             '/limits',
  896             body={'limits': [ref1, ref2]},
  897             token=self.system_admin_token,
  898             expected_status=http.client.CREATED)
  899         id1 = r.result['limits'][0]['id']
  900         r = self.get(
  901             '/limits',
  902             expected_status=http.client.OK)
  903         limits = r.result['limits']
  904         self.assertEqual(len(limits), 2)
  905         if limits[0]['id'] == id1:
  906             self.assertEqual(limits[0]['region_id'], ref1['region_id'])
  907             self.assertIsNone(limits[1].get('region_id'))
  908             for key in ['service_id', 'resource_name', 'resource_limit']:
  909                 self.assertEqual(limits[0][key], ref1[key])
  910                 self.assertEqual(limits[1][key], ref2[key])
  911         else:
  912             self.assertEqual(limits[1]['region_id'], ref1['region_id'])
  913             self.assertIsNone(limits[0].get('region_id'))
  914             for key in ['service_id', 'resource_name', 'resource_limit']:
  915                 self.assertEqual(limits[1][key], ref1[key])
  916                 self.assertEqual(limits[0][key], ref2[key])
  917 
  918         r = self.get(
  919             '/limits?service_id=%s' % self.service_id2,
  920             expected_status=http.client.OK)
  921         limits = r.result['limits']
  922         self.assertEqual(len(limits), 1)
  923         for key in ['service_id', 'resource_name', 'resource_limit']:
  924             self.assertEqual(limits[0][key], ref2[key])
  925 
  926         r = self.get(
  927             '/limits?region_id=%s' % self.region_id,
  928             expected_status=http.client.OK)
  929         limits = r.result['limits']
  930         self.assertEqual(len(limits), 1)
  931         for key in ['service_id', 'region_id', 'resource_name',
  932                     'resource_limit']:
  933             self.assertEqual(limits[0][key], ref1[key])
  934 
  935         r = self.get(
  936             '/limits?resource_name=volume',
  937             expected_status=http.client.OK)
  938         limits = r.result['limits']
  939         self.assertEqual(len(limits), 1)
  940         for key in ['service_id', 'region_id', 'resource_name',
  941                     'resource_limit']:
  942             self.assertEqual(limits[0][key], ref1[key])
  943 
  944     def test_list_limit_with_project_id_filter(self):
  945         # create two limit in different projects for test.
  946         self.config_fixture.config(group='oslo_policy',
  947                                    enforce_scope=True)
  948         ref1 = unit.new_limit_ref(project_id=self.project_id,
  949                                   service_id=self.service_id,
  950                                   region_id=self.region_id,
  951                                   resource_name='volume')
  952         ref2 = unit.new_limit_ref(project_id=self.project_2_id,
  953                                   service_id=self.service_id2,
  954                                   resource_name='snapshot')
  955         self.post(
  956             '/limits',
  957             body={'limits': [ref1, ref2]},
  958             token=self.system_admin_token,
  959             expected_status=http.client.CREATED)
  960 
  961         # non system scoped request will get the limits in its project.
  962         r = self.get('/limits', expected_status=http.client.OK)
  963         limits = r.result['limits']
  964         self.assertEqual(1, len(limits))
  965         self.assertEqual(self.project_id, limits[0]['project_id'])
  966 
  967         r = self.get(
  968             '/limits', expected_status=http.client.OK,
  969             auth=self.build_authentication_request(
  970                 user_id=self.user['id'], password=self.user['password'],
  971                 project_id=self.project_2_id))
  972         limits = r.result['limits']
  973         self.assertEqual(1, len(limits))
  974         self.assertEqual(self.project_2_id, limits[0]['project_id'])
  975 
  976         # any project user can filter by their own project
  977         r = self.get(
  978             '/limits?project_id=%s' % self.project_id,
  979             expected_status=http.client.OK)
  980         limits = r.result['limits']
  981         self.assertEqual(1, len(limits))
  982         self.assertEqual(self.project_id, limits[0]['project_id'])
  983 
  984         # a system scoped request can specify the project_id filter
  985         r = self.get(
  986             '/limits?project_id=%s' % self.project_id,
  987             expected_status=http.client.OK,
  988             token=self.system_admin_token
  989         )
  990         limits = r.result['limits']
  991         self.assertEqual(1, len(limits))
  992         self.assertEqual(self.project_id, limits[0]['project_id'])
  993 
  994     def test_list_limit_with_domain_id_filter(self):
  995         # create two limit in different domains for test.
  996         ref1 = unit.new_limit_ref(domain_id=self.domain_id,
  997                                   service_id=self.service_id,
  998                                   region_id=self.region_id,
  999                                   resource_name='volume')
 1000         ref2 = unit.new_limit_ref(domain_id=self.domain_2_id,
 1001                                   service_id=self.service_id2,
 1002                                   resource_name='snapshot')
 1003         self.post(
 1004             '/limits',
 1005             body={'limits': [ref1, ref2]},
 1006             token=self.system_admin_token,
 1007             expected_status=http.client.CREATED)
 1008 
 1009         # non system scoped request will get the limits in its domain.
 1010         r = self.get(
 1011             '/limits', expected_status=http.client.OK,
 1012             auth=self.build_authentication_request(
 1013                 user_id=self.user['id'], password=self.user['password'],
 1014                 domain_id=self.domain_id))
 1015         limits = r.result['limits']
 1016         self.assertEqual(1, len(limits))
 1017         self.assertEqual(self.domain_id, limits[0]['domain_id'])
 1018 
 1019         r = self.get(
 1020             '/limits', expected_status=http.client.OK,
 1021             auth=self.build_authentication_request(
 1022                 user_id=self.user['id'], password=self.user['password'],
 1023                 domain_id=self.domain_2_id))
 1024         limits = r.result['limits']
 1025         self.assertEqual(1, len(limits))
 1026         self.assertEqual(self.domain_2_id, limits[0]['domain_id'])
 1027 
 1028         # if non system scoped request contain domain_id filter, keystone
 1029         # will return an empty list.
 1030         r = self.get(
 1031             '/limits?domain_id=%s' % self.domain_id,
 1032             expected_status=http.client.OK)
 1033         limits = r.result['limits']
 1034         self.assertEqual(0, len(limits))
 1035 
 1036         # a system scoped request can specify the domain_id filter
 1037         r = self.get(
 1038             '/limits?domain_id=%s' % self.domain_id,
 1039             expected_status=http.client.OK,
 1040             auth=self.build_authentication_request(
 1041                 user_id=self.user['id'], password=self.user['password'],
 1042                 system=True))
 1043         limits = r.result['limits']
 1044         self.assertEqual(1, len(limits))
 1045         self.assertEqual(self.domain_id, limits[0]['domain_id'])
 1046 
 1047     def test_show_project_limit(self):
 1048         ref1 = unit.new_limit_ref(project_id=self.project_id,
 1049                                   service_id=self.service_id,
 1050                                   region_id=self.region_id,
 1051                                   resource_name='volume')
 1052         ref2 = unit.new_limit_ref(project_id=self.project_id,
 1053                                   service_id=self.service_id2,
 1054                                   resource_name='snapshot')
 1055         r = self.post(
 1056             '/limits',
 1057             body={'limits': [ref1, ref2]},
 1058             token=self.system_admin_token,
 1059             expected_status=http.client.CREATED)
 1060         if r.result['limits'][0]['resource_name'] == 'volume':
 1061             id1 = r.result['limits'][0]['id']
 1062         else:
 1063             id1 = r.result['limits'][1]['id']
 1064         self.get('/limits/fake_id',
 1065                  token=self.system_admin_token,
 1066                  expected_status=http.client.NOT_FOUND)
 1067         r = self.get('/limits/%s' % id1,
 1068                      expected_status=http.client.OK)
 1069         limit = r.result['limit']
 1070         self.assertIsNone(limit['domain_id'])
 1071         for key in ['service_id', 'region_id', 'resource_name',
 1072                     'resource_limit', 'description', 'project_id']:
 1073             self.assertEqual(limit[key], ref1[key])
 1074 
 1075     def test_show_domain_limit(self):
 1076         ref1 = unit.new_limit_ref(domain_id=self.domain_id,
 1077                                   service_id=self.service_id2,
 1078                                   resource_name='snapshot')
 1079         r = self.post(
 1080             '/limits',
 1081             body={'limits': [ref1]},
 1082             token=self.system_admin_token,
 1083             expected_status=http.client.CREATED)
 1084         id1 = r.result['limits'][0]['id']
 1085 
 1086         r = self.get('/limits/%s' % id1,
 1087                      expected_status=http.client.OK,
 1088                      auth=self.build_authentication_request(
 1089                          user_id=self.user['id'],
 1090                          password=self.user['password'],
 1091                          domain_id=self.domain_id))
 1092         limit = r.result['limit']
 1093         self.assertIsNone(limit['project_id'])
 1094         self.assertIsNone(limit['region_id'])
 1095         for key in ['service_id', 'resource_name', 'resource_limit',
 1096                     'description', 'domain_id']:
 1097             self.assertEqual(limit[key], ref1[key])
 1098 
 1099     def test_delete_limit(self):
 1100         ref1 = unit.new_limit_ref(project_id=self.project_id,
 1101                                   service_id=self.service_id,
 1102                                   region_id=self.region_id,
 1103                                   resource_name='volume')
 1104         ref2 = unit.new_limit_ref(project_id=self.project_id,
 1105                                   service_id=self.service_id2,
 1106                                   resource_name='snapshot')
 1107         r = self.post(
 1108             '/limits',
 1109             body={'limits': [ref1, ref2]},
 1110             token=self.system_admin_token,
 1111             expected_status=http.client.CREATED)
 1112         id1 = r.result['limits'][0]['id']
 1113         self.delete('/limits/%s' % id1,
 1114                     token=self.system_admin_token,
 1115                     expected_status=http.client.NO_CONTENT)
 1116         self.delete('/limits/fake_id',
 1117                     token=self.system_admin_token,
 1118                     expected_status=http.client.NOT_FOUND)
 1119         r = self.get(
 1120             '/limits',
 1121             token=self.system_admin_token,
 1122             expected_status=http.client.OK)
 1123         limits = r.result['limits']
 1124         self.assertEqual(len(limits), 1)
 1125 
 1126 
 1127 class StrictTwoLevelLimitsTestCase(LimitsTestCase):
 1128 
 1129     def setUp(self):
 1130         super(StrictTwoLevelLimitsTestCase, self).setUp()
 1131         # Most of these tests require system-scoped tokens. Let's have one on
 1132         # hand so that we can use it in tests when we need it.
 1133         PROVIDERS.assignment_api.create_system_grant_for_user(
 1134             self.user_id, self.role_id
 1135         )
 1136         self.system_admin_token = self.get_system_scoped_token()
 1137 
 1138         # create two hierarchical projects trees for test. The first level is
 1139         # domain.
 1140         #   A        D
 1141         #  / \      / \
 1142         # B   C    E   F
 1143         domain_ref = {'domain': {'name': 'A', 'enabled': True}}
 1144         response = self.post('/domains', body=domain_ref)
 1145         self.domain_A = response.json_body['domain']
 1146         project_ref = {'project': {'name': 'B', 'enabled': True,
 1147                                    'domain_id': self.domain_A['id']}}
 1148         response = self.post('/projects', body=project_ref)
 1149         self.project_B = response.json_body['project']
 1150         project_ref = {'project': {'name': 'C', 'enabled': True,
 1151                                    'domain_id': self.domain_A['id']}}
 1152         response = self.post('/projects', body=project_ref)
 1153         self.project_C = response.json_body['project']
 1154 
 1155         domain_ref = {'domain': {'name': 'D', 'enabled': True}}
 1156         response = self.post('/domains', body=domain_ref)
 1157         self.domain_D = response.json_body['domain']
 1158         project_ref = {'project': {'name': 'E', 'enabled': True,
 1159                                    'domain_id': self.domain_D['id']}}
 1160         response = self.post('/projects', body=project_ref)
 1161         self.project_E = response.json_body['project']
 1162         project_ref = {'project': {'name': 'F', 'enabled': True,
 1163                                    'domain_id': self.domain_D['id']}}
 1164         response = self.post('/projects', body=project_ref)
 1165         self.project_F = response.json_body['project']
 1166 
 1167     def config_overrides(self):
 1168         super(StrictTwoLevelLimitsTestCase, self).config_overrides()
 1169         self.config_fixture.config(group='unified_limit',
 1170                                    enforcement_model='strict_two_level')
 1171 
 1172     def test_create_child_limit(self):
 1173         # when A is 20, success to create B to 15, C to 18.
 1174         #     A,20             A,20
 1175         #    / \      -->     / \
 1176         #   B   C           B,15 C,18
 1177         ref = unit.new_limit_ref(domain_id=self.domain_A['id'],
 1178                                  service_id=self.service_id,
 1179                                  region_id=self.region_id,
 1180                                  resource_name='volume',
 1181                                  resource_limit=20)
 1182         self.post(
 1183             '/limits',
 1184             body={'limits': [ref]},
 1185             token=self.system_admin_token,
 1186             expected_status=http.client.CREATED)
 1187 
 1188         ref = unit.new_limit_ref(project_id=self.project_B['id'],
 1189                                  service_id=self.service_id,
 1190                                  region_id=self.region_id,
 1191                                  resource_name='volume',
 1192                                  resource_limit=15)
 1193         self.post(
 1194             '/limits',
 1195             body={'limits': [ref]},
 1196             token=self.system_admin_token,
 1197             expected_status=http.client.CREATED)
 1198 
 1199         ref = unit.new_limit_ref(project_id=self.project_C['id'],
 1200                                  service_id=self.service_id,
 1201                                  region_id=self.region_id,
 1202                                  resource_name='volume',
 1203                                  resource_limit=18)
 1204         self.post(
 1205             '/limits',
 1206             body={'limits': [ref]},
 1207             token=self.system_admin_token,
 1208             expected_status=http.client.CREATED)
 1209 
 1210     def test_create_child_limit_break_hierarchical_tree(self):
 1211         # when A is 20, success to create B to 15, but fail to create C to 21.
 1212         #     A,20             A,20
 1213         #    / \      -->     / \
 1214         #   B   C           B,15 C
 1215         #
 1216         #     A,20              A,20
 1217         #    / \      -/->      / \
 1218         #  B,15 C            B,15 C,21
 1219         ref = unit.new_limit_ref(domain_id=self.domain_A['id'],
 1220                                  service_id=self.service_id,
 1221                                  region_id=self.region_id,
 1222                                  resource_name='volume',
 1223                                  resource_limit=20)
 1224         self.post(
 1225             '/limits',
 1226             body={'limits': [ref]},
 1227             token=self.system_admin_token,
 1228             expected_status=http.client.CREATED)
 1229 
 1230         ref = unit.new_limit_ref(project_id=self.project_B['id'],
 1231                                  service_id=self.service_id,
 1232                                  region_id=self.region_id,
 1233                                  resource_name='volume',
 1234                                  resource_limit=15)
 1235         self.post(
 1236             '/limits',
 1237             body={'limits': [ref]},
 1238             token=self.system_admin_token,
 1239             expected_status=http.client.CREATED)
 1240 
 1241         ref = unit.new_limit_ref(project_id=self.project_C['id'],
 1242                                  service_id=self.service_id,
 1243                                  region_id=self.region_id,
 1244                                  resource_name='volume',
 1245                                  resource_limit=21)
 1246         self.post(
 1247             '/limits',
 1248             body={'limits': [ref]},
 1249             token=self.system_admin_token,
 1250             expected_status=http.client.FORBIDDEN)
 1251 
 1252     def test_create_child_with_default_parent(self):
 1253         # If A is not set, the default value is 10 (from registered limit).
 1254         # success to create B to 5, but fail to create C to 11.
 1255         #     A(10)             A(10)
 1256         #    / \      -->      / \
 1257         #   B   C            B,5  C
 1258         #
 1259         #     A(10)             A(10)
 1260         #    / \     -/->      / \
 1261         #   B,5   C          B,5  C,11
 1262         ref = unit.new_limit_ref(project_id=self.project_B['id'],
 1263                                  service_id=self.service_id,
 1264                                  region_id=self.region_id,
 1265                                  resource_name='volume',
 1266                                  resource_limit=5)
 1267         self.post(
 1268             '/limits',
 1269             body={'limits': [ref]},
 1270             token=self.system_admin_token,
 1271             expected_status=http.client.CREATED)
 1272 
 1273         ref = unit.new_limit_ref(project_id=self.project_C['id'],
 1274                                  service_id=self.service_id,
 1275                                  region_id=self.region_id,
 1276                                  resource_name='volume',
 1277                                  resource_limit=11)
 1278         self.post(
 1279             '/limits',
 1280             body={'limits': [ref]},
 1281             token=self.system_admin_token,
 1282             expected_status=http.client.FORBIDDEN)
 1283 
 1284     def test_create_parent_limit(self):
 1285         # When B is 9 , success to set A to 12
 1286         #     A              A,12
 1287         #    / \    -->      / \
 1288         #  B,9  C          B,9  C
 1289         ref = unit.new_limit_ref(project_id=self.project_B['id'],
 1290                                  service_id=self.service_id,
 1291                                  region_id=self.region_id,
 1292                                  resource_name='volume',
 1293                                  resource_limit=9)
 1294         self.post(
 1295             '/limits',
 1296             body={'limits': [ref]},
 1297             token=self.system_admin_token,
 1298             expected_status=http.client.CREATED)
 1299 
 1300         ref = unit.new_limit_ref(domain_id=self.domain_A['id'],
 1301                                  service_id=self.service_id,
 1302                                  region_id=self.region_id,
 1303                                  resource_name='volume',
 1304                                  resource_limit=12)
 1305         self.post(
 1306             '/limits',
 1307             body={'limits': [ref]},
 1308             token=self.system_admin_token,
 1309             expected_status=http.client.CREATED)
 1310 
 1311     def test_create_parent_limit_break_hierarchical_tree(self):
 1312         # When B is 9 , fail to set A to 8
 1313         #     A              A,8
 1314         #    / \    -/->     / \
 1315         #  B,9  C          B,9  C
 1316         ref = unit.new_limit_ref(project_id=self.project_B['id'],
 1317                                  service_id=self.service_id,
 1318                                  region_id=self.region_id,
 1319                                  resource_name='volume',
 1320                                  resource_limit=9)
 1321         self.post(
 1322             '/limits',
 1323             body={'limits': [ref]},
 1324             token=self.system_admin_token,
 1325             expected_status=http.client.CREATED)
 1326 
 1327         ref = unit.new_limit_ref(domain_id=self.domain_A['id'],
 1328                                  service_id=self.service_id,
 1329                                  region_id=self.region_id,
 1330                                  resource_name='volume',
 1331                                  resource_limit=8)
 1332         self.post(
 1333             '/limits',
 1334             body={'limits': [ref]},
 1335             token=self.system_admin_token,
 1336             expected_status=http.client.FORBIDDEN)
 1337 
 1338     def test_create_multi_limits(self):
 1339         # success to create a tree in one request like:
 1340         #    A,12         D,9
 1341         #    / \          / \
 1342         #  B,9  C,5    E,5   F,4
 1343         ref_A = unit.new_limit_ref(domain_id=self.domain_A['id'],
 1344                                    service_id=self.service_id,
 1345                                    region_id=self.region_id,
 1346                                    resource_name='volume',
 1347                                    resource_limit=12)
 1348         ref_B = unit.new_limit_ref(project_id=self.project_B['id'],
 1349                                    service_id=self.service_id,
 1350                                    region_id=self.region_id,
 1351                                    resource_name='volume',
 1352                                    resource_limit=9)
 1353         ref_C = unit.new_limit_ref(project_id=self.project_C['id'],
 1354                                    service_id=self.service_id,
 1355                                    region_id=self.region_id,
 1356                                    resource_name='volume',
 1357                                    resource_limit=5)
 1358         ref_D = unit.new_limit_ref(domain_id=self.domain_D['id'],
 1359                                    service_id=self.service_id,
 1360                                    region_id=self.region_id,
 1361                                    resource_name='volume',
 1362                                    resource_limit=9)
 1363         ref_E = unit.new_limit_ref(project_id=self.project_E['id'],
 1364                                    service_id=self.service_id,
 1365                                    region_id=self.region_id,
 1366                                    resource_name='volume',
 1367                                    resource_limit=5)
 1368         ref_F = unit.new_limit_ref(project_id=self.project_F['id'],
 1369                                    service_id=self.service_id,
 1370                                    region_id=self.region_id,
 1371                                    resource_name='volume',
 1372                                    resource_limit=4)
 1373         self.post(
 1374             '/limits',
 1375             body={'limits': [ref_A, ref_B, ref_C, ref_D, ref_E, ref_F]},
 1376             token=self.system_admin_token,
 1377             expected_status=http.client.CREATED)
 1378 
 1379     def test_create_multi_limits_invalid_input(self):
 1380         # fail to create a tree in one request like:
 1381         #    A,12         D,9
 1382         #    / \          / \
 1383         #  B,9  C,5    E,5   F,10
 1384         # because F will break the second limit tree.
 1385         ref_A = unit.new_limit_ref(domain_id=self.domain_A['id'],
 1386                                    service_id=self.service_id,
 1387                                    region_id=self.region_id,
 1388                                    resource_name='volume',
 1389                                    resource_limit=12)
 1390         ref_B = unit.new_limit_ref(project_id=self.project_B['id'],
 1391                                    service_id=self.service_id,
 1392                                    region_id=self.region_id,
 1393                                    resource_name='volume',
 1394                                    resource_limit=9)
 1395         ref_C = unit.new_limit_ref(project_id=self.project_C['id'],
 1396                                    service_id=self.service_id,
 1397                                    region_id=self.region_id,
 1398                                    resource_name='volume',
 1399                                    resource_limit=5)
 1400         ref_D = unit.new_limit_ref(domain_id=self.domain_D['id'],
 1401                                    service_id=self.service_id,
 1402                                    region_id=self.region_id,
 1403                                    resource_name='volume',
 1404                                    resource_limit=9)
 1405         ref_E = unit.new_limit_ref(project_id=self.project_E['id'],
 1406                                    service_id=self.service_id,
 1407                                    region_id=self.region_id,
 1408                                    resource_name='volume',
 1409                                    resource_limit=5)
 1410         ref_F = unit.new_limit_ref(project_id=self.project_F['id'],
 1411                                    service_id=self.service_id,
 1412                                    region_id=self.region_id,
 1413                                    resource_name='volume',
 1414                                    resource_limit=10)
 1415         self.post(
 1416             '/limits',
 1417             body={'limits': [ref_A, ref_B, ref_C, ref_D, ref_E, ref_F]},
 1418             token=self.system_admin_token,
 1419             expected_status=http.client.FORBIDDEN)
 1420 
 1421     def test_create_multi_limits_break_hierarchical_tree(self):
 1422         # when there is some hierarchical_trees already like:
 1423         #    A,12          D
 1424         #    / \          / \
 1425         #  B,9  C       E,5   F
 1426         # fail to set C to 5 and D to 4 in one request like:
 1427         #    A,12         D,4
 1428         #    / \          / \
 1429         #  B,9  C,5    E,5   F
 1430         # because D will break the second limit tree.
 1431         ref_A = unit.new_limit_ref(domain_id=self.domain_A['id'],
 1432                                    service_id=self.service_id,
 1433                                    region_id=self.region_id,
 1434                                    resource_name='volume',
 1435                                    resource_limit=12)
 1436         ref_B = unit.new_limit_ref(project_id=self.project_B['id'],
 1437                                    service_id=self.service_id,
 1438                                    region_id=self.region_id,
 1439                                    resource_name='volume',
 1440                                    resource_limit=9)
 1441         ref_E = unit.new_limit_ref(project_id=self.project_E['id'],
 1442                                    service_id=self.service_id,
 1443                                    region_id=self.region_id,
 1444                                    resource_name='volume',
 1445                                    resource_limit=5)
 1446         self.post(
 1447             '/limits',
 1448             body={'limits': [ref_A, ref_B, ref_E]},
 1449             token=self.system_admin_token,
 1450             expected_status=http.client.CREATED)
 1451 
 1452         ref_C = unit.new_limit_ref(project_id=self.project_C['id'],
 1453                                    service_id=self.service_id,
 1454                                    region_id=self.region_id,
 1455                                    resource_name='volume',
 1456                                    resource_limit=5)
 1457         ref_D = unit.new_limit_ref(domain_id=self.domain_D['id'],
 1458                                    service_id=self.service_id,
 1459                                    region_id=self.region_id,
 1460                                    resource_name='volume',
 1461                                    resource_limit=4)
 1462         self.post(
 1463             '/limits',
 1464             body={'limits': [ref_C, ref_D]},
 1465             token=self.system_admin_token,
 1466             expected_status=http.client.FORBIDDEN)
 1467 
 1468     def test_update_child_limit(self):
 1469         # Success to update C to 9
 1470         #     A,10             A,10
 1471         #    / \      -->     / \
 1472         #  B,6  C,7         B,6  C,9
 1473         ref_A = unit.new_limit_ref(domain_id=self.domain_A['id'],
 1474                                    service_id=self.service_id,
 1475                                    region_id=self.region_id,
 1476                                    resource_name='volume',
 1477                                    resource_limit=10)
 1478         ref_B = unit.new_limit_ref(project_id=self.project_B['id'],
 1479                                    service_id=self.service_id,
 1480                                    region_id=self.region_id,
 1481                                    resource_name='volume',
 1482                                    resource_limit=6)
 1483         ref_C = unit.new_limit_ref(project_id=self.project_C['id'],
 1484                                    service_id=self.service_id,
 1485                                    region_id=self.region_id,
 1486                                    resource_name='volume',
 1487                                    resource_limit=7)
 1488         self.post(
 1489             '/limits',
 1490             body={'limits': [ref_A, ref_B]},
 1491             token=self.system_admin_token,
 1492             expected_status=http.client.CREATED)
 1493         r = self.post(
 1494             '/limits',
 1495             body={'limits': [ref_C]},
 1496             token=self.system_admin_token,
 1497             expected_status=http.client.CREATED)
 1498 
 1499         update_dict = {'resource_limit': 9}
 1500         self.patch(
 1501             '/limits/%s' % r.result['limits'][0]['id'],
 1502             body={'limit': update_dict},
 1503             token=self.system_admin_token,
 1504             expected_status=http.client.OK)
 1505 
 1506     def test_update_child_limit_break_hierarchical_tree(self):
 1507         # Fail to update C to 11
 1508         #     A,10             A,10
 1509         #    / \      -/->     / \
 1510         #  B,6  C,7         B,6  C,11
 1511         ref_A = unit.new_limit_ref(domain_id=self.domain_A['id'],
 1512                                    service_id=self.service_id,
 1513                                    region_id=self.region_id,
 1514                                    resource_name='volume',
 1515                                    resource_limit=10)
 1516         ref_B = unit.new_limit_ref(project_id=self.project_B['id'],
 1517                                    service_id=self.service_id,
 1518                                    region_id=self.region_id,
 1519                                    resource_name='volume',
 1520                                    resource_limit=6)
 1521         ref_C = unit.new_limit_ref(project_id=self.project_C['id'],
 1522                                    service_id=self.service_id,
 1523                                    region_id=self.region_id,
 1524                                    resource_name='volume',
 1525                                    resource_limit=7)
 1526         self.post(
 1527             '/limits',
 1528             body={'limits': [ref_A, ref_B]},
 1529             token=self.system_admin_token,
 1530             expected_status=http.client.CREATED)
 1531         r = self.post(
 1532             '/limits',
 1533             body={'limits': [ref_C]},
 1534             token=self.system_admin_token,
 1535             expected_status=http.client.CREATED)
 1536 
 1537         update_dict = {'resource_limit': 11}
 1538         self.patch(
 1539             '/limits/%s' % r.result['limits'][0]['id'],
 1540             body={'limit': update_dict},
 1541             token=self.system_admin_token,
 1542             expected_status=http.client.FORBIDDEN)
 1543 
 1544     def test_update_child_limit_with_default_parent(self):
 1545         # If A is not set, the default value is 10 (from registered limit).
 1546         # Success to update C to 9 but fail to update C to 11
 1547         #     A,(10)           A,(10)
 1548         #    / \      -->     / \
 1549         #   B,  C,7          B  C,9
 1550         #
 1551         #     A,(10)            A,(10)
 1552         #    / \      -/->     / \
 1553         #   B,  C,7           B  C,11
 1554         ref_C = unit.new_limit_ref(project_id=self.project_C['id'],
 1555                                    service_id=self.service_id,
 1556                                    region_id=self.region_id,
 1557                                    resource_name='volume',
 1558                                    resource_limit=7)
 1559         r = self.post(
 1560             '/limits',
 1561             body={'limits': [ref_C]},
 1562             token=self.system_admin_token,
 1563             expected_status=http.client.CREATED)
 1564 
 1565         update_dict = {'resource_limit': 9}
 1566         self.patch(
 1567             '/limits/%s' % r.result['limits'][0]['id'],
 1568             body={'limit': update_dict},
 1569             token=self.system_admin_token,
 1570             expected_status=http.client.OK)
 1571 
 1572         update_dict = {'resource_limit': 11}
 1573         self.patch(
 1574             '/limits/%s' % r.result['limits'][0]['id'],
 1575             body={'limit': update_dict},
 1576             token=self.system_admin_token,
 1577             expected_status=http.client.FORBIDDEN)
 1578 
 1579     def test_update_parent_limit(self):
 1580         # Success to update A to 8
 1581         #     A,10             A,8
 1582         #    / \      -->     / \
 1583         #  B,6  C,7         B,6  C,7
 1584         ref_A = unit.new_limit_ref(domain_id=self.domain_A['id'],
 1585                                    service_id=self.service_id,
 1586                                    region_id=self.region_id,
 1587                                    resource_name='volume',
 1588                                    resource_limit=10)
 1589         ref_B = unit.new_limit_ref(project_id=self.project_B['id'],
 1590                                    service_id=self.service_id,
 1591                                    region_id=self.region_id,
 1592                                    resource_name='volume',
 1593                                    resource_limit=6)
 1594         ref_C = unit.new_limit_ref(project_id=self.project_C['id'],
 1595                                    service_id=self.service_id,
 1596                                    region_id=self.region_id,
 1597                                    resource_name='volume',
 1598                                    resource_limit=7)
 1599         r = self.post(
 1600             '/limits',
 1601             body={'limits': [ref_A]},
 1602             token=self.system_admin_token,
 1603             expected_status=http.client.CREATED)
 1604         self.post(
 1605             '/limits',
 1606             body={'limits': [ref_B, ref_C]},
 1607             token=self.system_admin_token,
 1608             expected_status=http.client.CREATED)
 1609 
 1610         update_dict = {'resource_limit': 8}
 1611         self.patch(
 1612             '/limits/%s' % r.result['limits'][0]['id'],
 1613             body={'limit': update_dict},
 1614             token=self.system_admin_token,
 1615             expected_status=http.client.OK)
 1616 
 1617     def test_update_parent_limit_break_hierarchical_tree(self):
 1618         # Fail to update A to 6
 1619         #     A,10             A,6
 1620         #    / \      -/->     / \
 1621         #  B,6  C,7         B,6  C,7
 1622         ref_A = unit.new_limit_ref(domain_id=self.domain_A['id'],
 1623                                    service_id=self.service_id,
 1624                                    region_id=self.region_id,
 1625                                    resource_name='volume',
 1626                                    resource_limit=10)
 1627         ref_B = unit.new_limit_ref(project_id=self.project_B['id'],
 1628                                    service_id=self.service_id,
 1629                                    region_id=self.region_id,
 1630                                    resource_name='volume',
 1631                                    resource_limit=6)
 1632         ref_C = unit.new_limit_ref(project_id=self.project_C['id'],
 1633                                    service_id=self.service_id,
 1634                                    region_id=self.region_id,
 1635                                    resource_name='volume',
 1636                                    resource_limit=7)
 1637         r = self.post(
 1638             '/limits',
 1639             body={'limits': [ref_A]},
 1640             token=self.system_admin_token,
 1641             expected_status=http.client.CREATED)
 1642         self.post(
 1643             '/limits',
 1644             body={'limits': [ref_B, ref_C]},
 1645             token=self.system_admin_token,
 1646             expected_status=http.client.CREATED)
 1647 
 1648         update_dict = {'resource_limit': 6}
 1649         self.patch(
 1650             '/limits/%s' % r.result['limits'][0]['id'],
 1651             body={'limit': update_dict},
 1652             token=self.system_admin_token,
 1653             expected_status=http.client.FORBIDDEN)