"Fossies" - the Fresh Open Source Software Archive

Member "keystone-17.0.0/keystone/tests/unit/filtering.py" (13 May 2020, 5085 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 "filtering.py": 16.0.1_vs_17.0.0.

    1 # Copyright 2013 OpenStack Foundation
    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 uuid
   16 
   17 from keystone.common import provider_api
   18 import keystone.conf
   19 from keystone import exception
   20 
   21 
   22 PROVIDERS = provider_api.ProviderAPIs
   23 CONF = keystone.conf.CONF
   24 
   25 
   26 class FilterTests(object):
   27 
   28     # Provide support for checking if a batch of list items all
   29     # exist within a contiguous range in a total list
   30     def _match_with_list(self, this_batch, total_list,
   31                          batch_size=None,
   32                          list_start=None, list_end=None):
   33         if batch_size is None:
   34             batch_size = len(this_batch)
   35         if list_start is None:
   36             list_start = 0
   37         if list_end is None:
   38             list_end = len(total_list)
   39         for batch_item in range(0, batch_size):
   40             found = False
   41             for list_item in range(list_start, list_end):
   42                 if this_batch[batch_item]['id'] == total_list[list_item]['id']:
   43                     found = True
   44             self.assertTrue(found)
   45 
   46     def _create_entity(self, entity_type):
   47         """Find the create_<entity_type> method.
   48 
   49         Searches through the [identity_api, resource_api, assignment_api]
   50         managers for a method called create_<entity_type> and returns the first
   51         one.
   52 
   53         """
   54         f = getattr(PROVIDERS.identity_api, 'create_%s' % entity_type, None)
   55         if f is None:
   56             f = getattr(
   57                 PROVIDERS.resource_api, 'create_%s' % entity_type, None
   58             )
   59         if f is None:
   60             f = getattr(PROVIDERS.assignment_api, 'create_%s' % entity_type)
   61         return f
   62 
   63     def _delete_entity(self, entity_type):
   64         """Find the delete_<entity_type> method.
   65 
   66         Searches through the [identity_api, resource_api, assignment_api]
   67         managers for a method called delete_<entity_type> and returns the first
   68         one.
   69 
   70         """
   71         f = getattr(PROVIDERS.identity_api, 'delete_%s' % entity_type, None)
   72         if f is None:
   73             f = getattr(
   74                 PROVIDERS.resource_api, 'delete_%s' % entity_type, None
   75             )
   76         if f is None:
   77             f = getattr(PROVIDERS.assignment_api, 'delete_%s' % entity_type)
   78         return f
   79 
   80     def _list_entities(self, entity_type):
   81         """Find the list_<entity_type> method.
   82 
   83         Searches through the [identity_api, resource_api, assignment_api]
   84         managers for a method called list_<entity_type> and returns the first
   85         one.
   86 
   87         """
   88         f = getattr(PROVIDERS.identity_api, 'list_%ss' % entity_type, None)
   89         if f is None:
   90             f = getattr(PROVIDERS.resource_api, 'list_%ss' % entity_type, None)
   91         if f is None:
   92             f = getattr(PROVIDERS.assignment_api, 'list_%ss' % entity_type)
   93         return f
   94 
   95     def _create_one_entity(self, entity_type, domain_id, name):
   96         new_entity = {'name': name,
   97                       'domain_id': domain_id}
   98         if entity_type in ['user', 'group']:
   99             # The manager layer creates the ID for users and groups
  100             new_entity = self._create_entity(entity_type)(new_entity)
  101         else:
  102             new_entity['id'] = uuid.uuid4().hex
  103             self._create_entity(entity_type)(new_entity['id'], new_entity)
  104         return new_entity
  105 
  106     def _create_test_data(self, entity_type, number, domain_id=None,
  107                           name_dict=None):
  108         """Create entity test data.
  109 
  110         :param entity_type: type of entity to create, e.g. 'user', group' etc.
  111         :param number: number of entities to create,
  112         :param domain_id: if not defined, all users will be created in the
  113                           default domain.
  114         :param name_dict: optional dict containing entity number and name pairs
  115 
  116         """
  117         entity_list = []
  118         if domain_id is None:
  119             domain_id = CONF.identity.default_domain_id
  120         name_dict = name_dict or {}
  121         for x in range(number):
  122             # If this index has a name defined in the name_dict, then use it
  123             name = name_dict.get(x, uuid.uuid4().hex)
  124             new_entity = self._create_one_entity(entity_type, domain_id, name)
  125             entity_list.append(new_entity)
  126         return entity_list
  127 
  128     def _delete_test_data(self, entity_type, entity_list):
  129         for entity in entity_list:
  130             try:
  131                 self._delete_entity(entity_type)(entity['id'])
  132             except exception.Forbidden:
  133                 # Note(knikolla): Some identity backends such as LDAP are
  134                 # read only
  135                 break