1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Copyright 2012 OpenStack LLC
5
# Licensed under the Apache License, Version 2.0 (the "License"); you may
6
# not use this file except in compliance with the License. You may obtain
7
# a copy of the License at
9
# http://www.apache.org/licenses/LICENSE-2.0
11
# Unless required by applicable law or agreed to in writing, software
12
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
13
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14
# License for the specific language governing permissions and limitations
20
from migrate.versioning import api as versioning_api
23
from keystone.common import sql
24
from keystone import config
25
from keystone import test
26
from keystone.common.sql import migration
27
import default_fixtures
32
class SqlUpgradeTests(test.TestCase):
34
super(SqlUpgradeTests, self).setUp()
35
self.config([test.etcdir('keystone.conf.sample'),
36
test.testsdir('test_overrides.conf'),
37
test.testsdir('backend_sql.conf')])
39
# create and share a single sqlalchemy engine for testing
40
self.engine = sql.Base().get_engine(allow_global_engine=False)
41
self.metadata = sqlalchemy.MetaData()
43
# populate the engine with tables & fixtures
44
self.metadata.bind = self.engine
45
self.repo_path = migration._find_migrate_repo()
46
self.schema = versioning_api.ControlledSchema.create(self.engine,
50
super(SqlUpgradeTests, self).tearDown()
52
def test_blank_db_to_start(self):
53
self.assertTableDoesNotExist('user')
55
def test_start_version_0(self):
56
version = migration.db_version()
57
self.assertEqual(version, 0, "DB is at version 0")
59
def assertTableColumns(self, table_name, expected_cols):
60
"""Asserts that the table contains the expected set of columns."""
61
table = self.select_table(table_name)
62
actual_cols = [col.name for col in table.columns]
63
self.assertEqual(expected_cols, actual_cols, '%s table' % table_name)
65
def test_upgrade_0_to_1(self):
66
self.assertEqual(self.schema.version, 0, "DB is at version 0")
67
self._migrate(self.repo_path, 1)
68
self.assertEqual(self.schema.version, 1, "DB is at version 1")
69
self.assertTableColumns("user", ["id", "name", "extra"])
70
self.assertTableColumns("tenant", ["id", "name", "extra"])
71
self.assertTableColumns("role", ["id", "name"])
72
self.assertTableColumns("user_tenant_membership",
73
["user_id", "tenant_id"])
74
self.assertTableColumns("metadata", ["user_id", "tenant_id", "data"])
75
self.populate_user_table()
77
def test_upgrade_5_to_6(self):
78
self._migrate(self.repo_path, 5)
79
self.assertEqual(self.schema.version, 5)
80
self.assertTableDoesNotExist('policy')
82
self._migrate(self.repo_path, 6)
83
self.assertEqual(self.schema.version, 6)
84
self.assertTableExists('policy')
85
self.assertTableColumns('policy', ['id', 'type', 'blob', 'extra'])
87
def populate_user_table(self):
88
for user in default_fixtures.USERS:
89
extra = copy.deepcopy(user)
92
self.engine.execute("insert into user values ('%s', '%s', '%s')"
97
def select_table(self, name):
98
table = sqlalchemy.Table(name,
101
s = sqlalchemy.select([table])
104
def assertTableExists(self, table_name):
105
"""Asserts that a given table exists can be selected by name."""
107
self.select_table(table_name)
108
except sqlalchemy.exc.NoSuchTableError:
109
raise AssertionError('Table "%s" does not exist' % table_name)
111
def assertTableDoesNotExist(self, table_name):
112
"""Asserts that a given table exists cannot be selected by name."""
114
self.assertTableExists(table_name)
115
except AssertionError:
118
raise AssertionError('Table "%s" already exists' % table_name)
120
def _migrate(self, repository, version):
123
version = versioning_api._migrate_version(self.schema,
127
changeset = self.schema.changeset(version)
128
for ver, change in changeset:
129
self.schema.runchange(ver, change, changeset.step)
added
removed
tests/test_sql_upgrade.py
