"Fossies" - the Fresh Open Source Software Archive

Member "keystone-17.0.0/keystone/tests/unit/identity/test_core.py" (13 May 2020, 7749 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_core.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 """Unit tests for core identity behavior."""
   14 
   15 import itertools
   16 import os
   17 from unittest import mock
   18 import uuid
   19 
   20 from oslo_config import fixture as config_fixture
   21 
   22 from keystone.common import provider_api
   23 import keystone.conf
   24 from keystone import exception
   25 from keystone import identity
   26 from keystone.tests import unit
   27 from keystone.tests.unit import default_fixtures
   28 from keystone.tests.unit.ksfixtures import database
   29 
   30 
   31 CONF = keystone.conf.CONF
   32 PROVIDERS = provider_api.ProviderAPIs
   33 
   34 
   35 class TestDomainConfigs(unit.BaseTestCase):
   36 
   37     def setUp(self):
   38         super(TestDomainConfigs, self).setUp()
   39         self.addCleanup(CONF.reset)
   40 
   41         self.tmp_dir = unit.dirs.tmp()
   42 
   43         self.config_fixture = self.useFixture(config_fixture.Config(CONF))
   44         self.config_fixture.config(domain_config_dir=self.tmp_dir,
   45                                    group='identity')
   46 
   47     def test_config_for_nonexistent_domain(self):
   48         """Having a config for a non-existent domain will be ignored.
   49 
   50         There are no assertions in this test because there are no side
   51         effects. If there is a config file for a domain that does not
   52         exist it should be ignored.
   53 
   54         """
   55         domain_id = uuid.uuid4().hex
   56         domain_config_filename = os.path.join(self.tmp_dir,
   57                                               'keystone.%s.conf' % domain_id)
   58         self.addCleanup(lambda: os.remove(domain_config_filename))
   59         with open(domain_config_filename, 'w'):
   60             """Write an empty config file."""
   61 
   62         e = exception.DomainNotFound(domain_id=domain_id)
   63         mock_assignment_api = mock.Mock()
   64         mock_assignment_api.get_domain_by_name.side_effect = e
   65 
   66         domain_config = identity.DomainConfigs()
   67         fake_standard_driver = None
   68         domain_config.setup_domain_drivers(fake_standard_driver,
   69                                            mock_assignment_api)
   70 
   71     def test_config_for_dot_name_domain(self):
   72         # Ensure we can get the right domain name which has dots within it
   73         # from filename.
   74         domain_config_filename = os.path.join(self.tmp_dir,
   75                                               'keystone.abc.def.com.conf')
   76         with open(domain_config_filename, 'w'):
   77             """Write an empty config file."""
   78         self.addCleanup(os.remove, domain_config_filename)
   79 
   80         with mock.patch.object(identity.DomainConfigs,
   81                                '_load_config_from_file') as mock_load_config:
   82             domain_config = identity.DomainConfigs()
   83             fake_assignment_api = None
   84             fake_standard_driver = None
   85             domain_config.setup_domain_drivers(fake_standard_driver,
   86                                                fake_assignment_api)
   87             mock_load_config.assert_called_once_with(fake_assignment_api,
   88                                                      [domain_config_filename],
   89                                                      'abc.def.com')
   90 
   91     def test_config_for_multiple_sql_backend(self):
   92         domains_config = identity.DomainConfigs()
   93 
   94         # Create the right sequence of is_sql in the drivers being
   95         # requested to expose the bug, which is that a False setting
   96         # means it forgets previous True settings.
   97         drivers = []
   98         files = []
   99         for idx, is_sql in enumerate((True, False, True)):
  100             drv = mock.Mock(is_sql=is_sql)
  101             drivers.append(drv)
  102             name = 'dummy.{0}'.format(idx)
  103             files.append(''.join((
  104                 identity.DOMAIN_CONF_FHEAD,
  105                 name,
  106                 identity.DOMAIN_CONF_FTAIL)))
  107 
  108         def walk_fake(*a, **kwa):
  109             return ('/fake/keystone/domains/config', [], files),
  110 
  111         generic_driver = mock.Mock(is_sql=False)
  112 
  113         assignment_api = mock.Mock()
  114         id_factory = itertools.count()
  115         assignment_api.get_domain_by_name.side_effect = (
  116             lambda name: {'id': next(id_factory), '_': 'fake_domain'})
  117         load_driver_mock = mock.Mock(side_effect=drivers)
  118 
  119         with mock.patch.object(os, 'walk', walk_fake):
  120             with mock.patch.object(identity.cfg, 'ConfigOpts'):
  121                 with mock.patch.object(domains_config, '_load_driver',
  122                                        load_driver_mock):
  123                     self.assertRaises(
  124                         exception.MultipleSQLDriversInConfig,
  125                         domains_config.setup_domain_drivers,
  126                         generic_driver, assignment_api)
  127 
  128                     self.assertEqual(3, load_driver_mock.call_count)
  129 
  130 
  131 class TestDatabaseDomainConfigs(unit.TestCase):
  132 
  133     def setUp(self):
  134         super(TestDatabaseDomainConfigs, self).setUp()
  135         self.useFixture(database.Database())
  136         self.load_backends()
  137         PROVIDERS.resource_api.create_domain(
  138             default_fixtures.ROOT_DOMAIN['id'], default_fixtures.ROOT_DOMAIN)
  139 
  140     def test_domain_config_in_database_disabled_by_default(self):
  141         self.assertFalse(CONF.identity.domain_configurations_from_database)
  142 
  143     def test_loading_config_from_database(self):
  144         self.config_fixture.config(domain_configurations_from_database=True,
  145                                    group='identity')
  146         domain = unit.new_domain_ref()
  147         PROVIDERS.resource_api.create_domain(domain['id'], domain)
  148         # Override two config options for our domain
  149         conf = {'ldap': {'url': uuid.uuid4().hex,
  150                          'suffix': uuid.uuid4().hex,
  151                          'use_tls': True},
  152                 'identity': {
  153                     'driver': 'ldap'}}
  154         PROVIDERS.domain_config_api.create_config(domain['id'], conf)
  155         fake_standard_driver = None
  156         domain_config = identity.DomainConfigs()
  157         domain_config.setup_domain_drivers(fake_standard_driver,
  158                                            PROVIDERS.resource_api)
  159         # Make sure our two overrides are in place, and others are not affected
  160         res = domain_config.get_domain_conf(domain['id'])
  161         self.assertEqual(conf['ldap']['url'], res.ldap.url)
  162         self.assertEqual(conf['ldap']['suffix'], res.ldap.suffix)
  163         self.assertEqual(CONF.ldap.query_scope, res.ldap.query_scope)
  164 
  165         # Make sure the override is not changing the type of the config value
  166         use_tls_type = type(CONF.ldap.use_tls)
  167         self.assertEqual(use_tls_type(conf['ldap']['use_tls']),
  168                          res.ldap.use_tls)
  169 
  170         # Now turn off using database domain configuration and check that the
  171         # default config file values are now seen instead of the overrides.
  172         self.config_fixture.config(
  173             group='identity',
  174             domain_configurations_from_database=False)
  175 
  176         domain_config = identity.DomainConfigs()
  177         domain_config.setup_domain_drivers(fake_standard_driver,
  178                                            PROVIDERS.resource_api)
  179         res = domain_config.get_domain_conf(domain['id'])
  180         self.assertEqual(CONF.ldap.url, res.ldap.url)
  181         self.assertEqual(CONF.ldap.suffix, res.ldap.suffix)
  182         self.assertEqual(CONF.ldap.use_tls, res.ldap.use_tls)
  183         self.assertEqual(CONF.ldap.query_scope, res.ldap.query_scope)