"Fossies" - the Fresh Open Source Software Archive

Member "keystone-17.0.0/keystone/tests/protection/v3/test_limits.py" (13 May 2020, 28602 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 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    2 #    not use this file except in compliance with the License. You may obtain
    3 #    a copy of the License at
    4 #
    5 #         http://www.apache.org/licenses/LICENSE-2.0
    6 #
    7 #    Unless required by applicable law or agreed to in writing, software
    8 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    9 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   10 #    License for the specific language governing permissions and limitations
   11 #    under the License.
   12 
   13 import uuid
   14 
   15 import http.client
   16 
   17 from keystone.common import provider_api
   18 import keystone.conf
   19 from keystone.tests.common import auth as common_auth
   20 from keystone.tests import unit
   21 from keystone.tests.unit import base_classes
   22 from keystone.tests.unit import ksfixtures
   23 
   24 CONF = keystone.conf.CONF
   25 PROVIDERS = provider_api.ProviderAPIs
   26 
   27 
   28 def _create_limits_and_dependencies(domain_id=None):
   29     """Create limits and its dependencies for testing."""
   30     if not domain_id:
   31         domain_id = CONF.identity.default_domain_id
   32 
   33     service = PROVIDERS.catalog_api.create_service(
   34         uuid.uuid4().hex, unit.new_service_ref()
   35     )
   36 
   37     registered_limit = unit.new_registered_limit_ref(
   38         service_id=service['id'], id=uuid.uuid4().hex
   39     )
   40     registered_limits = (
   41         PROVIDERS.unified_limit_api.create_registered_limits(
   42             [registered_limit]
   43         )
   44     )
   45     registered_limit = registered_limits[0]
   46 
   47     domain_limit = unit.new_limit_ref(
   48         domain_id=domain_id, service_id=service['id'],
   49         resource_name=registered_limit['resource_name'],
   50         resource_limit=10, id=uuid.uuid4().hex
   51     )
   52 
   53     project = PROVIDERS.resource_api.create_project(
   54         uuid.uuid4().hex, unit.new_project_ref(domain_id=domain_id)
   55     )
   56 
   57     project_limit = unit.new_limit_ref(
   58         project_id=project['id'], service_id=service['id'],
   59         resource_name=registered_limit['resource_name'],
   60         resource_limit=5, id=uuid.uuid4().hex
   61     )
   62     limits = PROVIDERS.unified_limit_api.create_limits(
   63         [domain_limit, project_limit]
   64     )
   65 
   66     project_limit_id = None
   67     domain_limit_id = None
   68     for limit in limits:
   69         if limit.get('domain_id'):
   70             domain_limit_id = limit['id']
   71         else:
   72             project_limit_id = limit['id']
   73 
   74     return (project_limit_id, domain_limit_id)
   75 
   76 
   77 class _UserLimitTests(object):
   78     """Common default functionality for all users except system admins."""
   79 
   80     def test_user_can_get_limit_model(self):
   81         with self.test_client() as c:
   82             c.get('/v3/limits/model', headers=self.headers)
   83 
   84     def test_user_can_get_a_limit(self):
   85         limit_id, _ = _create_limits_and_dependencies()
   86 
   87         with self.test_client() as c:
   88             r = c.get('/v3/limits/%s' % limit_id, headers=self.headers)
   89             self.assertEqual(limit_id, r.json['limit']['id'])
   90 
   91     def test_user_can_list_limits(self):
   92         project_limit_id, domain_limit_id = _create_limits_and_dependencies()
   93 
   94         with self.test_client() as c:
   95             r = c.get('/v3/limits', headers=self.headers)
   96             self.assertTrue(len(r.json['limits']) == 2)
   97             result = []
   98             for limit in r.json['limits']:
   99                 result.append(limit['id'])
  100 
  101             self.assertIn(project_limit_id, result)
  102             self.assertIn(domain_limit_id, result)
  103 
  104     def test_user_cannot_create_limits(self):
  105         service = PROVIDERS.catalog_api.create_service(
  106             uuid.uuid4().hex, unit.new_service_ref()
  107         )
  108 
  109         registered_limit = unit.new_registered_limit_ref(
  110             service_id=service['id'], id=uuid.uuid4().hex
  111         )
  112         registered_limits = (
  113             PROVIDERS.unified_limit_api.create_registered_limits(
  114                 [registered_limit]
  115             )
  116         )
  117         registered_limit = registered_limits[0]
  118 
  119         project = PROVIDERS.resource_api.create_project(
  120             uuid.uuid4().hex,
  121             unit.new_project_ref(domain_id=CONF.identity.default_domain_id)
  122         )
  123 
  124         create = {
  125             'limits': [
  126                 unit.new_limit_ref(
  127                     project_id=project['id'], service_id=service['id'],
  128                     resource_name=registered_limit['resource_name'],
  129                     resource_limit=5
  130                 )
  131             ]
  132         }
  133 
  134         with self.test_client() as c:
  135             c.post(
  136                 '/v3/limits', json=create, headers=self.headers,
  137                 expected_status_code=http.client.FORBIDDEN
  138             )
  139 
  140     def test_user_cannot_update_limits(self):
  141         limit_id, _ = _create_limits_and_dependencies()
  142 
  143         update = {'limits': {'description': uuid.uuid4().hex}}
  144 
  145         with self.test_client() as c:
  146             c.patch(
  147                 '/v3/limits/%s' % limit_id, json=update,
  148                 headers=self.headers,
  149                 expected_status_code=http.client.FORBIDDEN
  150             )
  151 
  152     def test_user_cannot_delete_limits(self):
  153         limit_id, _ = _create_limits_and_dependencies()
  154 
  155         with self.test_client() as c:
  156             c.delete(
  157                 '/v3/limits/%s' % limit_id,
  158                 headers=self.headers,
  159                 expected_status_code=http.client.FORBIDDEN
  160             )
  161 
  162 
  163 class SystemReaderTests(base_classes.TestCaseWithBootstrap,
  164                         common_auth.AuthTestMixin,
  165                         _UserLimitTests):
  166     def setUp(self):
  167         super(SystemReaderTests, self).setUp()
  168         self.loadapp()
  169         self.useFixture(ksfixtures.Policy(self.config_fixture))
  170         self.config_fixture.config(group='oslo_policy', enforce_scope=True)
  171 
  172         system_reader = unit.new_user_ref(
  173             domain_id=CONF.identity.default_domain_id
  174         )
  175         self.user_id = PROVIDERS.identity_api.create_user(
  176             system_reader
  177         )['id']
  178         PROVIDERS.assignment_api.create_system_grant_for_user(
  179             self.user_id, self.bootstrapper.reader_role_id
  180         )
  181 
  182         auth = self.build_authentication_request(
  183             user_id=self.user_id, password=system_reader['password'],
  184             system=True
  185         )
  186 
  187         # Grab a token using the persona we're testing and prepare headers
  188         # for requests we'll be making in the tests.
  189         with self.test_client() as c:
  190             r = c.post('/v3/auth/tokens', json=auth)
  191             self.token_id = r.headers['X-Subject-Token']
  192             self.headers = {'X-Auth-Token': self.token_id}
  193 
  194 
  195 class SystemMemberTests(base_classes.TestCaseWithBootstrap,
  196                         common_auth.AuthTestMixin,
  197                         _UserLimitTests):
  198 
  199     def setUp(self):
  200         super(SystemMemberTests, self).setUp()
  201         self.loadapp()
  202         self.useFixture(ksfixtures.Policy(self.config_fixture))
  203         self.config_fixture.config(group='oslo_policy', enforce_scope=True)
  204 
  205         system_member = unit.new_user_ref(
  206             domain_id=CONF.identity.default_domain_id
  207         )
  208         self.user_id = PROVIDERS.identity_api.create_user(
  209             system_member
  210         )['id']
  211         PROVIDERS.assignment_api.create_system_grant_for_user(
  212             self.user_id, self.bootstrapper.member_role_id
  213         )
  214 
  215         auth = self.build_authentication_request(
  216             user_id=self.user_id, password=system_member['password'],
  217             system=True
  218         )
  219 
  220         # Grab a token using the persona we're testing and prepare headers
  221         # for requests we'll be making in the tests.
  222         with self.test_client() as c:
  223             r = c.post('/v3/auth/tokens', json=auth)
  224             self.token_id = r.headers['X-Subject-Token']
  225             self.headers = {'X-Auth-Token': self.token_id}
  226 
  227 
  228 class SystemAdminTests(base_classes.TestCaseWithBootstrap,
  229                        common_auth.AuthTestMixin):
  230 
  231     def setUp(self):
  232         super(SystemAdminTests, self).setUp()
  233         self.loadapp()
  234         self.useFixture(ksfixtures.Policy(self.config_fixture))
  235         self.config_fixture.config(group='oslo_policy', enforce_scope=True)
  236 
  237         # Reuse the system administrator account created during
  238         # ``keystone-manage bootstrap``
  239         self.user_id = self.bootstrapper.admin_user_id
  240         auth = self.build_authentication_request(
  241             user_id=self.user_id,
  242             password=self.bootstrapper.admin_password,
  243             system=True
  244         )
  245 
  246         # Grab a token using the persona we're testing and prepare headers
  247         # for requests we'll be making in the tests.
  248         with self.test_client() as c:
  249             r = c.post('/v3/auth/tokens', json=auth)
  250             self.token_id = r.headers['X-Subject-Token']
  251             self.headers = {'X-Auth-Token': self.token_id}
  252 
  253     def test_user_can_get_a_limit(self):
  254         limit_id, _ = _create_limits_and_dependencies()
  255 
  256         with self.test_client() as c:
  257             r = c.get('/v3/limits/%s' % limit_id, headers=self.headers)
  258             self.assertEqual(limit_id, r.json['limit']['id'])
  259 
  260     def test_user_can_list_limits(self):
  261         project_limit_id, domain_limit_id = _create_limits_and_dependencies()
  262 
  263         with self.test_client() as c:
  264             r = c.get('/v3/limits', headers=self.headers)
  265             self.assertTrue(len(r.json['limits']) == 2)
  266             result = []
  267             for limit in r.json['limits']:
  268                 result.append(limit['id'])
  269 
  270             self.assertIn(project_limit_id, result)
  271             self.assertIn(domain_limit_id, result)
  272 
  273     def test_user_can_create_limits(self):
  274         service = PROVIDERS.catalog_api.create_service(
  275             uuid.uuid4().hex, unit.new_service_ref()
  276         )
  277 
  278         registered_limit = unit.new_registered_limit_ref(
  279             service_id=service['id'], id=uuid.uuid4().hex
  280         )
  281         registered_limits = (
  282             PROVIDERS.unified_limit_api.create_registered_limits(
  283                 [registered_limit]
  284             )
  285         )
  286         registered_limit = registered_limits[0]
  287 
  288         project = PROVIDERS.resource_api.create_project(
  289             uuid.uuid4().hex,
  290             unit.new_project_ref(domain_id=CONF.identity.default_domain_id)
  291         )
  292 
  293         create = {
  294             'limits': [
  295                 unit.new_limit_ref(
  296                     project_id=project['id'], service_id=service['id'],
  297                     resource_name=registered_limit['resource_name'],
  298                     resource_limit=5
  299                 )
  300             ]
  301         }
  302 
  303         with self.test_client() as c:
  304             c.post('/v3/limits', json=create, headers=self.headers)
  305 
  306     def test_user_can_update_limits(self):
  307         limit_id, _ = _create_limits_and_dependencies()
  308 
  309         update = {'limits': {'description': uuid.uuid4().hex}}
  310 
  311         with self.test_client() as c:
  312             c.patch(
  313                 '/v3/limits/%s' % limit_id, json=update,
  314                 headers=self.headers
  315             )
  316 
  317     def test_user_can_delete_limits(self):
  318         limit_id, _ = _create_limits_and_dependencies()
  319 
  320         with self.test_client() as c:
  321             c.delete('/v3/limits/%s' % limit_id, headers=self.headers)
  322 
  323 
  324 class DomainUserTests(base_classes.TestCaseWithBootstrap,
  325                       common_auth.AuthTestMixin):
  326 
  327     def setUp(self):
  328         super(DomainUserTests, self).setUp()
  329         self.loadapp()
  330         self.useFixture(ksfixtures.Policy(self.config_fixture))
  331         self.config_fixture.config(group='oslo_policy', enforce_scope=True)
  332 
  333         domain = PROVIDERS.resource_api.create_domain(
  334             uuid.uuid4().hex, unit.new_domain_ref()
  335         )
  336         self.domain_id = domain['id']
  337         domain_admin = unit.new_user_ref(domain_id=self.domain_id)
  338         self.user_id = PROVIDERS.identity_api.create_user(domain_admin)['id']
  339         PROVIDERS.assignment_api.create_grant(
  340             self.bootstrapper.admin_role_id, user_id=self.user_id,
  341             domain_id=self.domain_id
  342         )
  343 
  344         auth = self.build_authentication_request(
  345             user_id=self.user_id,
  346             password=domain_admin['password'],
  347             domain_id=self.domain_id
  348         )
  349 
  350         # Grab a token using the persona we're testing and prepare headers
  351         # for requests we'll be making in the tests.
  352         with self.test_client() as c:
  353             r = c.post('/v3/auth/tokens', json=auth)
  354             self.token_id = r.headers['X-Subject-Token']
  355             self.headers = {'X-Auth-Token': self.token_id}
  356 
  357     def test_user_can_get_project_limits_within_domain(self):
  358         project_limit_id, _ = _create_limits_and_dependencies(
  359             domain_id=self.domain_id
  360         )
  361 
  362         with self.test_client() as c:
  363             c.get('/v3/limits/%s' % project_limit_id, headers=self.headers)
  364 
  365     def test_user_can_get_domain_limits(self):
  366         _, domain_limit_id = _create_limits_and_dependencies(
  367             domain_id=self.domain_id
  368         )
  369 
  370         with self.test_client() as c:
  371             r = c.get('/v3/limits/%s' % domain_limit_id, headers=self.headers)
  372             self.assertEqual(self.domain_id, r.json['limit']['domain_id'])
  373 
  374     def test_user_cannot_get_project_limit_outside_domain(self):
  375         project_limit_id, _ = _create_limits_and_dependencies()
  376 
  377         with self.test_client() as c:
  378             c.get(
  379                 '/v3/limits/%s' % project_limit_id, headers=self.headers,
  380                 expected_status_code=http.client.FORBIDDEN
  381             )
  382 
  383     def test_user_cannot_get_domain_limits_for_other_domain(self):
  384         _, domain_limit_id = _create_limits_and_dependencies()
  385 
  386         with self.test_client() as c:
  387             c.get(
  388                 '/v3/limits/%s' % domain_limit_id, headers=self.headers,
  389                 expected_status_code=http.client.FORBIDDEN
  390             )
  391 
  392     def test_user_can_list_limits_within_domain(self):
  393         project_limit_id, domain_limit_id = _create_limits_and_dependencies(
  394             domain_id=self.domain_id
  395         )
  396 
  397         with self.test_client() as c:
  398             r = c.get('/v3/limits', headers=self.headers)
  399             result = []
  400             for limit in r.json['limits']:
  401                 result.append(limit['id'])
  402 
  403             self.assertEqual(2, len(r.json['limits']))
  404             self.assertIn(project_limit_id, result)
  405             self.assertIn(domain_limit_id, result)
  406 
  407     def test_user_cannot_list_limits_outside_domain(self):
  408         _create_limits_and_dependencies()
  409 
  410         with self.test_client() as c:
  411             r = c.get('/v3/limits', headers=self.headers)
  412             self.assertEqual(0, len(r.json['limits']))
  413 
  414     def test_user_cannot_create_limits_for_domain(self):
  415         service = PROVIDERS.catalog_api.create_service(
  416             uuid.uuid4().hex, unit.new_service_ref()
  417         )
  418 
  419         registered_limit = unit.new_registered_limit_ref(
  420             service_id=service['id'], id=uuid.uuid4().hex
  421         )
  422         registered_limits = (
  423             PROVIDERS.unified_limit_api.create_registered_limits(
  424                 [registered_limit]
  425             )
  426         )
  427         registered_limit = registered_limits[0]
  428 
  429         create = {
  430             'limits': [
  431                 unit.new_limit_ref(
  432                     domain_id=self.domain_id, service_id=service['id'],
  433                     resource_name=registered_limit['resource_name'],
  434                     resource_limit=5
  435                 )
  436             ]
  437         }
  438 
  439         with self.test_client() as c:
  440             c.post(
  441                 '/v3/limits', json=create, headers=self.headers,
  442                 expected_status_code=http.client.FORBIDDEN
  443             )
  444 
  445     def test_user_cannot_create_limits_for_other_domain(self):
  446         service = PROVIDERS.catalog_api.create_service(
  447             uuid.uuid4().hex, unit.new_service_ref()
  448         )
  449 
  450         registered_limit = unit.new_registered_limit_ref(
  451             service_id=service['id'], id=uuid.uuid4().hex
  452         )
  453         registered_limits = (
  454             PROVIDERS.unified_limit_api.create_registered_limits(
  455                 [registered_limit]
  456             )
  457         )
  458         registered_limit = registered_limits[0]
  459 
  460         create = {
  461             'limits': [
  462                 unit.new_limit_ref(
  463                     domain_id=CONF.identity.default_domain_id,
  464                     service_id=service['id'],
  465                     resource_name=registered_limit['resource_name'],
  466                     resource_limit=5
  467                 )
  468             ]
  469         }
  470 
  471         with self.test_client() as c:
  472             c.post(
  473                 '/v3/limits', json=create, headers=self.headers,
  474                 expected_status_code=http.client.FORBIDDEN
  475             )
  476 
  477     def test_user_cannot_create_limits_for_projects_in_domain(self):
  478         service = PROVIDERS.catalog_api.create_service(
  479             uuid.uuid4().hex, unit.new_service_ref()
  480         )
  481 
  482         registered_limit = unit.new_registered_limit_ref(
  483             service_id=service['id'], id=uuid.uuid4().hex
  484         )
  485         registered_limits = (
  486             PROVIDERS.unified_limit_api.create_registered_limits(
  487                 [registered_limit]
  488             )
  489         )
  490         registered_limit = registered_limits[0]
  491 
  492         project = PROVIDERS.resource_api.create_project(
  493             uuid.uuid4().hex, unit.new_project_ref(domain_id=self.domain_id)
  494         )
  495 
  496         create = {
  497             'limits': [
  498                 unit.new_limit_ref(
  499                     project_id=project['id'],
  500                     service_id=service['id'],
  501                     resource_name=registered_limit['resource_name'],
  502                     resource_limit=5
  503                 )
  504             ]
  505         }
  506 
  507         with self.test_client() as c:
  508             c.post(
  509                 '/v3/limits', json=create, headers=self.headers,
  510                 expected_status_code=http.client.FORBIDDEN
  511             )
  512 
  513     def test_user_cannot_create_limits_for_projects_outside_domain(self):
  514         service = PROVIDERS.catalog_api.create_service(
  515             uuid.uuid4().hex, unit.new_service_ref()
  516         )
  517 
  518         registered_limit = unit.new_registered_limit_ref(
  519             service_id=service['id'], id=uuid.uuid4().hex
  520         )
  521         registered_limits = (
  522             PROVIDERS.unified_limit_api.create_registered_limits(
  523                 [registered_limit]
  524             )
  525         )
  526         registered_limit = registered_limits[0]
  527 
  528         project = PROVIDERS.resource_api.create_project(
  529             uuid.uuid4().hex,
  530             unit.new_project_ref(domain_id=CONF.identity.default_domain_id)
  531         )
  532 
  533         create = {
  534             'limits': [
  535                 unit.new_limit_ref(
  536                     project_id=project['id'],
  537                     service_id=service['id'],
  538                     resource_name=registered_limit['resource_name'],
  539                     resource_limit=5
  540                 )
  541             ]
  542         }
  543 
  544         with self.test_client() as c:
  545             c.post(
  546                 '/v3/limits', json=create, headers=self.headers,
  547                 expected_status_code=http.client.FORBIDDEN
  548             )
  549 
  550     def test_user_cannot_update_limits_for_domain(self):
  551         _, domain_limit_id = _create_limits_and_dependencies(
  552             domain_id=self.domain_id
  553         )
  554 
  555         update = {'limit': {'resource_limit': 1}}
  556 
  557         with self.test_client() as c:
  558             c.patch(
  559                 '/v3/limits/%s' % domain_limit_id, json=update,
  560                 headers=self.headers,
  561                 expected_status_code=http.client.FORBIDDEN
  562             )
  563 
  564     def test_user_cannot_update_limits_for_other_domain(self):
  565         _, domain_limit_id = _create_limits_and_dependencies()
  566 
  567         update = {'limit': {'resource_limit': 1}}
  568 
  569         with self.test_client() as c:
  570             c.patch(
  571                 '/v3/limits/%s' % domain_limit_id, json=update,
  572                 headers=self.headers,
  573                 expected_status_code=http.client.FORBIDDEN
  574             )
  575 
  576     def test_user_cannot_update_limits_for_projects_in_domain(self):
  577         project_limit_id, _ = _create_limits_and_dependencies(
  578             domain_id=self.domain_id
  579         )
  580 
  581         update = {'limit': {'resource_limit': 1}}
  582 
  583         with self.test_client() as c:
  584             c.patch(
  585                 '/v3/limits/%s' % project_limit_id, headers=self.headers,
  586                 json=update, expected_status_code=http.client.FORBIDDEN
  587             )
  588 
  589     def test_user_cannot_update_limits_for_projects_outside_domain(self):
  590         project_limit_id, _ = _create_limits_and_dependencies()
  591 
  592         update = {'limit': {'resource_limit': 1}}
  593 
  594         with self.test_client() as c:
  595             c.patch(
  596                 '/v3/limits/%s' % project_limit_id, headers=self.headers,
  597                 json=update, expected_status_code=http.client.FORBIDDEN
  598             )
  599 
  600     def test_user_cannot_delete_limits_for_domain(self):
  601         _, domain_limit_id = _create_limits_and_dependencies(
  602             domain_id=self.domain_id
  603         )
  604 
  605         with self.test_client() as c:
  606             c.delete(
  607                 '/v3/limits/%s' % domain_limit_id, headers=self.headers,
  608                 expected_status_code=http.client.FORBIDDEN
  609             )
  610 
  611     def test_user_cannot_delete_limits_for_other_domain(self):
  612         _, domain_limit_id = _create_limits_and_dependencies()
  613 
  614         with self.test_client() as c:
  615             c.delete(
  616                 '/v3/limits/%s' % domain_limit_id, headers=self.headers,
  617                 expected_status_code=http.client.FORBIDDEN
  618             )
  619 
  620     def test_user_cannot_delete_limits_for_projects_in_domain(self):
  621         project_limit_id, _ = _create_limits_and_dependencies(
  622             domain_id=self.domain_id
  623         )
  624 
  625         with self.test_client() as c:
  626             c.delete(
  627                 '/v3/limits/%s' % project_limit_id, headers=self.headers,
  628                 expected_status_code=http.client.FORBIDDEN
  629             )
  630 
  631     def test_user_cannot_delete_limits_for_projects_outside_domain(self):
  632         project_limit_id, _ = _create_limits_and_dependencies()
  633 
  634         with self.test_client() as c:
  635             c.delete(
  636                 '/v3/limits/%s' % project_limit_id, headers=self.headers,
  637                 expected_status_code=http.client.FORBIDDEN
  638             )
  639 
  640 
  641 class ProjectUserTests(base_classes.TestCaseWithBootstrap,
  642                        common_auth.AuthTestMixin):
  643 
  644     def setUp(self):
  645         super(ProjectUserTests, self).setUp()
  646         self.loadapp()
  647         self.useFixture(ksfixtures.Policy(self.config_fixture))
  648         self.config_fixture.config(group='oslo_policy', enforce_scope=True)
  649 
  650         # Reuse the system administrator account created during
  651         # ``keystone-manage bootstrap``
  652         self.user_id = self.bootstrapper.admin_user_id
  653         auth = self.build_authentication_request(
  654             user_id=self.user_id,
  655             password=self.bootstrapper.admin_password,
  656             project_id=self.bootstrapper.project_id
  657         )
  658 
  659         # Grab a token using the persona we're testing and prepare headers
  660         # for requests we'll be making in the tests.
  661         with self.test_client() as c:
  662             r = c.post('/v3/auth/tokens', json=auth)
  663             self.token_id = r.headers['X-Subject-Token']
  664             self.headers = {'X-Auth-Token': self.token_id}
  665 
  666     def test_user_can_get_project_limit(self):
  667         project_limit_id, _ = _create_limits_and_dependencies()
  668 
  669         limit = PROVIDERS.unified_limit_api.get_limit(project_limit_id)
  670         # NOTE(lbragstad): Project users are only allowed to list limits for a
  671         # project if they actually have a role assignment on the project and
  672         # call the API with a project-scoped token.
  673         PROVIDERS.assignment_api.create_grant(
  674             self.bootstrapper.reader_role_id, user_id=self.user_id,
  675             project_id=limit['project_id']
  676         )
  677         auth = self.build_authentication_request(
  678             user_id=self.user_id,
  679             password=self.bootstrapper.admin_password,
  680             project_id=limit['project_id']
  681         )
  682         with self.test_client() as c:
  683             r = c.post('/v3/auth/tokens', json=auth)
  684             token_id = r.headers['X-Subject-Token']
  685             headers = {'X-Auth-Token': token_id}
  686 
  687         with self.test_client() as c:
  688             r = c.get('/v3/limits/%s' % project_limit_id, headers=headers)
  689 
  690     def test_user_cannot_get_project_limit_without_role_assignment(self):
  691         project_limit_id, _ = _create_limits_and_dependencies()
  692 
  693         with self.test_client() as c:
  694             c.get(
  695                 '/v3/limits/%s' % project_limit_id, headers=self.headers,
  696                 expected_status_code=http.client.FORBIDDEN
  697             )
  698 
  699     def test_user_cannot_get_domain_limit(self):
  700         _, domain_limit_id = _create_limits_and_dependencies()
  701 
  702         with self.test_client() as c:
  703             c.get(
  704                 '/v3/limits/%s' % domain_limit_id, headers=self.headers,
  705                 expected_status_code=http.client.FORBIDDEN
  706             )
  707 
  708     def test_user_can_list_limits(self):
  709         project_limit_id, _ = _create_limits_and_dependencies()
  710 
  711         limit = PROVIDERS.unified_limit_api.get_limit(project_limit_id)
  712         # NOTE(lbragstad): Project users are only allowed to list limits for a
  713         # project if they actually have a role assignment on the project and
  714         # call the API with a project-scoped token.
  715         PROVIDERS.assignment_api.create_grant(
  716             self.bootstrapper.reader_role_id, user_id=self.user_id,
  717             project_id=limit['project_id']
  718         )
  719         auth = self.build_authentication_request(
  720             user_id=self.user_id,
  721             password=self.bootstrapper.admin_password,
  722             project_id=limit['project_id']
  723         )
  724         with self.test_client() as c:
  725             r = c.post('/v3/auth/tokens', json=auth)
  726             token_id = r.headers['X-Subject-Token']
  727             headers = {'X-Auth-Token': token_id}
  728 
  729         with self.test_client() as c:
  730             r = c.get('/v3/limits', headers=headers)
  731             self.assertTrue(len(r.json['limits']) == 1)
  732             self.assertEqual(project_limit_id, r.json['limits'][0]['id'])
  733 
  734     def test_user_cannot_list_limits_without_project_role_assignment(self):
  735         _create_limits_and_dependencies()
  736 
  737         with self.test_client() as c:
  738             r = c.get('/v3/limits', headers=self.headers)
  739             self.assertEqual(0, len(r.json['limits']))
  740 
  741     def test_user_can_get_limit_model(self):
  742         with self.test_client() as c:
  743             c.get('/v3/limits/model', headers=self.headers)
  744 
  745     def test_user_cannot_create_limits(self):
  746         service = PROVIDERS.catalog_api.create_service(
  747             uuid.uuid4().hex, unit.new_service_ref()
  748         )
  749 
  750         registered_limit = unit.new_registered_limit_ref(
  751             service_id=service['id'], id=uuid.uuid4().hex
  752         )
  753         registered_limits = (
  754             PROVIDERS.unified_limit_api.create_registered_limits(
  755                 [registered_limit]
  756             )
  757         )
  758         registered_limit = registered_limits[0]
  759 
  760         project = PROVIDERS.resource_api.create_project(
  761             uuid.uuid4().hex,
  762             unit.new_project_ref(domain_id=CONF.identity.default_domain_id)
  763         )
  764 
  765         create = {
  766             'limits': [
  767                 unit.new_limit_ref(
  768                     project_id=project['id'], service_id=service['id'],
  769                     resource_name=registered_limit['resource_name'],
  770                     resource_limit=5
  771                 )
  772             ]
  773         }
  774 
  775         with self.test_client() as c:
  776             c.post(
  777                 '/v3/limits', json=create, headers=self.headers,
  778                 expected_status_code=http.client.FORBIDDEN
  779             )
  780 
  781     def test_user_cannot_update_limits(self):
  782         limit_id, _ = _create_limits_and_dependencies()
  783 
  784         update = {'limits': {'description': uuid.uuid4().hex}}
  785 
  786         with self.test_client() as c:
  787             c.patch(
  788                 '/v3/limits/%s' % limit_id, json=update,
  789                 headers=self.headers,
  790                 expected_status_code=http.client.FORBIDDEN
  791             )
  792 
  793     def test_user_cannot_delete_limits(self):
  794         limit_id, _ = _create_limits_and_dependencies()
  795 
  796         with self.test_client() as c:
  797             c.delete(
  798                 '/v3/limits/%s' % limit_id,
  799                 headers=self.headers,
  800                 expected_status_code=http.client.FORBIDDEN
  801             )
  802 
  803 
  804 class ProjectUserTestsWithoutEnforceScope(ProjectUserTests):
  805 
  806     def setUp(self):
  807         super(ProjectUserTestsWithoutEnforceScope, self).setUp()
  808         self.config_fixture.config(group='oslo_policy', enforce_scope=False)