openstack
/
keystone
Code Issues Proposed changes

Force SQLite to properly deal with foreign keys 

This will help with testing since SQLite will start enforcing the
foreign key relationships.

We will still have a problem with migrations for tables that refer to
each other. SQLite can't alter tables and sqlalchemy-migrates tmp table
strategy for migrations fails in this situation.

This patch did:
1. Add FK support for the tests. Disable it by default.
2. Make sure the Fk is disabled for test_sql_upgrade and
identity.backens.test_sql

Partial-Bug: #1744195

Co-Authored-By: wangxiyuan<wangxiyuan@huawei.com>
Change-Id: I276af7c0125dc2cb2c54215d54491665db1caa22
This commit is contained in:
David Stanek 2014-10-03 18:22:41 +00:00 committed by Morgan Fainberg
parent fae0296ab2
commit ed2b65a5dd
15 changed files with 91 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
keystone/common/sql/core.py
View File

@ -256,6 +256,14 @@ def _get_main_context_manager():
return _main_context_manager
# Now this function is only used for testing FK with sqlite.
def enable_sqlite_foreign_key():
global _main_context_manager
if not _main_context_manager:
_main_context_manager = enginefacade.transaction_context()
_main_context_manager.configure(sqlite_fk=True)
def cleanup():
global _main_context_manager

9
keystone/tests/unit/core.py
View File

@ -48,6 +48,7 @@ import keystone.conf
from keystone import exception
from keystone.identity.backends.ldap import common as ks_ldap
from keystone import notifications
from keystone.resource.backends import base as resource_base
from keystone.tests.unit import ksfixtures
from keystone.version import controllers
from keystone.version import service
@ -721,7 +722,7 @@ class TestCase(BaseTestCase):
provider_api.ProviderAPIs._clear_registry_instances()
self.useFixture(ksfixtures.BackendLoader(self))
def load_fixtures(self, fixtures):
def load_fixtures(self, fixtures, enable_sqlite_foreign_key=False):
"""Hacky basic and naive fixture loading based on a python module.
Expects that the various APIs into the various services are already
@ -737,6 +738,12 @@ class TestCase(BaseTestCase):
if (hasattr(self, 'identity_api') and
hasattr(self, 'assignment_api') and
hasattr(self, 'resource_api')):
# TODO(wxy): Once all test enable FKs, remove
# ``enable_sqlite_foreign_key`` and create the root domain by
# default.
if enable_sqlite_foreign_key:
self.resource_api.create_domain(resource_base.NULL_DOMAIN_ID,
fixtures.ROOT_DOMAIN)
for domain in fixtures.DOMAINS:
rv = PROVIDERS.resource_api.create_domain(domain['id'], domain)
attrname = 'domain_%s' % domain['id']

7
keystone/tests/unit/default_fixtures.py
View File

@ -159,6 +159,13 @@ ROLE_ASSIGNMENTS = [
},
]
# TODO(wxy): We should add the root domain ``<<keystone.domain.root>>`` as well
# when the FKs is enabled for the test. Merge ROOT_DOMAIN into DOMAINS once all
# test enable FKs.
ROOT_DOMAIN = {'enabled': True,
'id': '<<keystone.domain.root>>',
'name': '<<keystone.domain.root>>'}
DOMAINS = [{'description':
(u'The default domain'),
'enabled': True,

6
keystone/tests/unit/identity/backends/test_sql.py
View File

@ -30,8 +30,10 @@ class TestIdentityDriver(db_test.DbTestCase,
def setUp(self):
super(TestIdentityDriver, self).setUp()
# Set keystone's connection URL to be the test engine's url.
database.initialize_sql_session(self.engine.url)
# Set keystone's connection URL to be the test engine's url. Close
# sqlite FK to avoid conflicting with sql upgrade test.
database.initialize_sql_session(self.engine.url,
enforce_sqlite_fks=False)
# Override keystone's context manager to be oslo.db's global context
# manager.

15
keystone/tests/unit/ksfixtures/database.py
View File

@ -16,6 +16,7 @@ import os
import fixtures
from oslo_db import options as db_options
from oslo_db.sqlalchemy import enginefacade
from keystone.common import sql
import keystone.conf
@ -42,7 +43,8 @@ def run_once(f):
# NOTE(I159): Every execution all the options will be cleared. The method must
# be called at the every fixture initialization.
def initialize_sql_session(connection_str=unit.IN_MEM_DB_CONN_STRING):
def initialize_sql_session(connection_str=unit.IN_MEM_DB_CONN_STRING,
enforce_sqlite_fks=True):
# Make sure the DB is located in the correct location, in this case set
# the default value, as this should be able to be overridden in some
# test cases.
@ -50,6 +52,13 @@ def initialize_sql_session(connection_str=unit.IN_MEM_DB_CONN_STRING):
CONF,
connection=connection_str)
# Enable the Sqlite FKs for global engine by default.
facade = enginefacade.get_legacy_facade()
engine = facade.get_engine()
f_key = 'ON' if enforce_sqlite_fks else 'OFF'
if engine.name == 'sqlite':
engine.connect().execute('PRAGMA foreign_keys = ' + f_key)
@run_once
def _load_sqlalchemy_models():
@ -94,10 +103,12 @@ def _load_sqlalchemy_models():
class Database(fixtures.Fixture):
"""A fixture for setting up and tearing down a database."""
def __init__(self):
def __init__(self, enable_sqlite_foreign_key=False):
super(Database, self).__init__()
initialize_sql_session()
_load_sqlalchemy_models()
if enable_sqlite_foreign_key:
sql.enable_sqlite_foreign_key()
def setUp(self):
super(Database, self).setUp()

8
keystone/tests/unit/rest.py
View File

@ -52,14 +52,16 @@ class RestfulTestCase(unit.TestCase):
# default content type to test
content_type = 'json'
def setUp(self, app_conf='keystone'):
def setUp(self, app_conf='keystone', enable_sqlite_foreign_key=False):
super(RestfulTestCase, self).setUp()
self.auth_plugin_config_override()
self.useFixture(database.Database())
self.useFixture(database.Database(
enable_sqlite_foreign_key=enable_sqlite_foreign_key))
self.load_backends()
self.load_fixtures(default_fixtures)
self.load_fixtures(default_fixtures,
enable_sqlite_foreign_key=enable_sqlite_foreign_key)
self.public_app = webtest.TestApp(
self.loadapp(app_conf, name='main'))

6
keystone/tests/unit/test_backend_ldap.py
View File

@ -1891,7 +1891,7 @@ class LDAPIdentityEnabledEmulation(LDAPIdentity, unit.TestCase):
super(LDAPIdentityEnabledEmulation, self).setUp()
_assert_backends(self, identity='ldap')
def load_fixtures(self, fixtures):
def load_fixtures(self, fixtures, enable_sqlite_foreign_key=False):
# Override super impl since need to create group container.
super(LDAPIdentity, self).load_fixtures(fixtures)
for obj in [self.tenant_bar, self.tenant_baz, self.user_foo,
@ -2310,7 +2310,7 @@ class MultiLDAPandSQLIdentity(BaseLDAPIdentity, unit.SQLDriverOverrides,
"""
def load_fixtures(self, fixtures):
def load_fixtures(self, fixtures, enable_sqlite_foreign_key=False):
self.domain_count = 5
self.domain_specific_count = 3
self.setup_initial_domains()
@ -2884,7 +2884,7 @@ class DomainSpecificLDAPandSQLIdentity(
super(DomainSpecificLDAPandSQLIdentity, self).setUp()
def load_fixtures(self, fixtures):
def load_fixtures(self, fixtures, enable_sqlite_foreign_key=False):
self.setup_initial_domains()
super(DomainSpecificLDAPandSQLIdentity, self).load_fixtures(fixtures)

6
keystone/tests/unit/test_sql_upgrade.py
View File

@ -205,8 +205,10 @@ class SqlMigrateBase(test_base.DbTestCase):
# modules have the same name (001_awesome.py).
self.addCleanup(script.PythonScript.clear)
# Set keystone's connection URL to be the test engine's url.
database.initialize_sql_session(self.engine.url)
# NOTE(dstanek): SQLAlchemy's migrate makes some assumptions in the
# SQLite driver about the lack of foreign key enforcement.
database.initialize_sql_session(self.engine.url,
enforce_sqlite_fks=False)
# Override keystone's context manager to be oslo.db's global context
# manager.

33
keystone/tests/unit/test_v3.py
View File

@ -28,6 +28,7 @@ from keystone.common import provider_api
from keystone.common.validation import validators
from keystone import exception
from keystone import middleware
from keystone.resource.backends import base as resource_base
from keystone.tests.common import auth as common_auth
from keystone.tests import unit
from keystone.tests.unit import rest
@ -200,9 +201,11 @@ class RestfulTestCase(unit.SQLDriverOverrides, rest.RestfulTestCase,
config_files.append(unit.dirs.tests_conf('backend_sql.conf'))
return config_files
def setUp(self, app_conf='keystone'):
def setUp(self, app_conf='keystone', enable_sqlite_foreign_key=False):
"""Setup for v3 Restful Test Cases."""
super(RestfulTestCase, self).setUp(app_conf=app_conf)
super(RestfulTestCase, self).setUp(
app_conf=app_conf,
enable_sqlite_foreign_key=enable_sqlite_foreign_key)
self.empty_context = {'environment': {}}
@ -212,21 +215,37 @@ class RestfulTestCase(unit.SQLDriverOverrides, rest.RestfulTestCase,
super(RestfulTestCase, self).load_backends()
def load_fixtures(self, fixtures):
self.load_sample_data()
def load_fixtures(self, fixtures, enable_sqlite_foreign_key=False):
self.load_sample_data(
enable_sqlite_foreign_key=enable_sqlite_foreign_key)
def _populate_default_domain(self):
def _populate_default_domain(self, enable_sqlite_foreign_key=False):
try:
PROVIDERS.resource_api.get_domain(DEFAULT_DOMAIN_ID)
except exception.DomainNotFound:
# TODO(wxy): when FK is enabled in sqlite, a lot of tests will fail
# due to the root domain is missing. So we should open FKs for the
# tests one by one. If the FKs is enabled for one test,
# `enable_sqlite_foreign_key` should be `true` here to ensure the
# root domain is created. Once all tests enable FKs, the
# ``enable_sqlite_foreign_key`` can be removed.
if enable_sqlite_foreign_key:
root_domain = unit.new_domain_ref(
id=resource_base.NULL_DOMAIN_ID,
name=resource_base.NULL_DOMAIN_ID
)
self.resource_api.create_domain(resource_base.NULL_DOMAIN_ID,
root_domain)
domain = unit.new_domain_ref(
description=(u'The default domain'),
id=DEFAULT_DOMAIN_ID,
name=u'Default')
PROVIDERS.resource_api.create_domain(DEFAULT_DOMAIN_ID, domain)
def load_sample_data(self, create_region_and_endpoints=True):
self._populate_default_domain()
def load_sample_data(self, create_region_and_endpoints=True,
enable_sqlite_foreign_key=False):
self._populate_default_domain(
enable_sqlite_foreign_key=enable_sqlite_foreign_key)
self.domain = unit.new_domain_ref()
self.domain_id = self.domain['id']
PROVIDERS.resource_api.create_domain(self.domain_id, self.domain)

4
keystone/tests/unit/test_v3_assignment.py
View File

@ -1154,7 +1154,7 @@ class RoleAssignmentBaseTestCase(test_v3.RestfulTestCase,
MAX_HIERARCHY_BREADTH = 3
MAX_HIERARCHY_DEPTH = CONF.max_project_tree_depth - 1
def load_sample_data(self):
def load_sample_data(self, enable_sqlite_foreign_key=False):
"""Create sample data to be used on tests.
Created data are i) a role and ii) a domain containing: a project
@ -3218,7 +3218,7 @@ class DomainSpecificRoleTests(test_v3.RestfulTestCase, unit.TestCase):
class ListUserProjectsTestCase(test_v3.RestfulTestCase):
"""Test for /users/<user>/projects."""
def load_sample_data(self):
def load_sample_data(self, enable_sqlite_foreign_key=False):
# do not load base class's data, keep it focused on the tests
self.auths = []

2
keystone/tests/unit/test_v3_auth.py
View File

@ -2626,7 +2626,7 @@ class TestFernetTokenAPIs(test_v3.RestfulTestCase, TokenAPITests,
class TestTokenRevokeSelfAndAdmin(test_v3.RestfulTestCase):
"""Test token revoke using v3 Identity API by token owner and admin."""
def load_sample_data(self):
def load_sample_data(self, enable_sqlite_foreign_key=False):
"""Load Sample Data for Test Cases.
Two domains, domainA and domainB

2
keystone/tests/unit/test_v3_catalog.py
View File

@ -998,7 +998,7 @@ class TestCatalogAPITemplatedProject(test_v3.RestfulTestCase):
super(TestCatalogAPITemplatedProject, self).config_overrides()
self.config_fixture.config(group='catalog', driver='templated')
def load_fixtures(self, fixtures):
def load_fixtures(self, fixtures, enable_sqlite_foreign_key=False):
self.load_sample_data(create_region_and_endpoints=False)
def test_project_delete(self):

8
keystone/tests/unit/test_v3_federation.py
View File

@ -1887,7 +1887,7 @@ class FederatedTokenTests(test_v3.RestfulTestCase, FederatedSetupMixin):
self.assertEqual(note['protocol'], protocol)
self.assertTrue(note['send_notification_called'])
def load_fixtures(self, fixtures):
def load_fixtures(self, fixtures, enable_sqlite_foreign_key=False):
super(FederatedTokenTests, self).load_fixtures(fixtures)
self.load_federation_sample_data()
@ -2828,7 +2828,7 @@ class FederatedTokenTests(test_v3.RestfulTestCase, FederatedSetupMixin):
class FernetFederatedTokenTests(test_v3.RestfulTestCase, FederatedSetupMixin):
AUTH_METHOD = 'token'
def load_fixtures(self, fixtures):
def load_fixtures(self, fixtures, enable_sqlite_foreign_key=False):
super(FernetFederatedTokenTests, self).load_fixtures(fixtures)
self.load_federation_sample_data()
@ -2944,7 +2944,7 @@ class FederatedUserTests(test_v3.RestfulTestCase, FederatedSetupMixin):
methods = ['saml2', 'token']
super(FederatedUserTests, self).auth_plugin_config_override(methods)
def load_fixtures(self, fixtures):
def load_fixtures(self, fixtures, enable_sqlite_foreign_key=False):
super(FederatedUserTests, self).load_fixtures(fixtures)
self.load_federation_sample_data()
@ -3313,7 +3313,7 @@ class ShadowMappingTests(test_v3.RestfulTestCase, FederatedSetupMixin):
methods = ['saml2', 'token']
super(ShadowMappingTests, self).auth_plugin_config_override(methods)
def load_fixtures(self, fixtures):
def load_fixtures(self, fixtures, enable_sqlite_foreign_key=False):
super(ShadowMappingTests, self).load_fixtures(fixtures)
self.load_federation_sample_data()

4
keystone/tests/unit/test_v3_filters.py
View File

@ -47,7 +47,7 @@ class IdentityTestFilteredCase(filtering.FilterTests,
self.tmpfilename = self.tempfile.file_name
super(IdentityTestFilteredCase, self).setUp()
def load_sample_data(self):
def load_sample_data(self, enable_sqlite_foreign_key=False):
"""Create sample data for these tests.
As well as the usual housekeeping, create a set of domains,
@ -341,7 +341,7 @@ class IdentityPasswordExpiryFilteredTestCase(filtering.FilterTests,
self.config_fixture = self.useFixture(config_fixture.Config(CONF))
super(IdentityPasswordExpiryFilteredTestCase, self).setUp()
def load_sample_data(self):
def load_sample_data(self, enable_sqlite_foreign_key=False):
"""Create sample data for password expiry tests.
The test environment will consist of a single domain, containing

6
keystone/tests/unit/test_v3_protection.py
View File

@ -64,7 +64,7 @@ class IdentityTestProtectedCase(test_v3.RestfulTestCase):
user_id=self.user1['id'],
password=self.user1['password'])
def load_sample_data(self):
def load_sample_data(self, enable_sqlite_foreign_key=False):
self._populate_default_domain()
# Start by creating a couple of domains
@ -395,7 +395,7 @@ class IdentityTestProtectedCase(test_v3.RestfulTestCase):
class IdentityTestPolicySample(test_v3.RestfulTestCase):
"""Test policy enforcement of the policy.json file."""
def load_sample_data(self):
def load_sample_data(self, enable_sqlite_foreign_key=False):
self._populate_default_domain()
self.just_a_user = unit.create_user(
@ -687,7 +687,7 @@ class IdentityTestv3CloudPolicySample(test_v3.RestfulTestCase,
)
)
def load_sample_data(self):
def load_sample_data(self, enable_sqlite_foreign_key=False):
# Start by creating a couple of domains
self._populate_default_domain()
self.domainA = unit.new_domain_ref()