From 3f503faac1fc3045364cb0d78ab7bb823739beff Mon Sep 17 00:00:00 2001 From: Roman Podolyaka Date: Thu, 30 May 2013 11:21:39 +0300 Subject: Add a monkey-patching util for sqlalchemy-migrate Nova and other projects use sqlalchemy-migrate for DB schema migrations. Unfortunately, this project looks like to be dead, but have some important bugs which makes lives of OpenStack developers harder (e. g. creation of a new unique constraint in SQLite leads to deletion of all existing unique constraints). Nova has some workarounds for bugs and limitations of sqlalchemy-migrate, though it would be nice to have those directly in sqlalchemy-migrate (at least in form of a monkey-patch for now). Oslo seems to be a good place to store this monkey-patch, so Nova and other projects could reuse it. This patch: - makes it possible to use the unified drop_unique_constraint() function for SQLite backend - fixes a bug in sqlalchemy-migrate that leads to deletion of existing unique constraints of a table when a new one is added (SQLite backend) Blueprint: oslo-sqlalchemy-migrate-uc-fixes Change-Id: Ifac07abac3814b3ea4dea5840b17a711f4b24b8d --- tests/unit/db/sqlalchemy/test_migrate.py | 95 ++++++++++++++++++++++++++++++++ 1 file changed, 95 insertions(+) create mode 100644 tests/unit/db/sqlalchemy/test_migrate.py (limited to 'tests') diff --git a/tests/unit/db/sqlalchemy/test_migrate.py b/tests/unit/db/sqlalchemy/test_migrate.py new file mode 100644 index 0000000..6724b5c --- /dev/null +++ b/tests/unit/db/sqlalchemy/test_migrate.py @@ -0,0 +1,95 @@ +# Copyright (c) 2013 OpenStack Foundation +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + + +from migrate.changeset.constraint import UniqueConstraint +from migrate.changeset.databases import sqlite +import sqlalchemy as sa + +from openstack.common.db.sqlalchemy import migration +from openstack.common.db.sqlalchemy import session +from tests.unit.db.sqlalchemy import base as test_base + + +def uniques(*constraints): + """Make a sequence of UniqueConstraint instances easily comparable + + Convert a sequence of UniqueConstraint instances into a set of + tuples of form (constraint_name, (constraint_columns)) so that + assertEquals() will be able to compare sets of unique constraints + + """ + + return set((uc.name, tuple(uc.columns.keys())) for uc in constraints) + + +class TestSqliteUniqueConstraints(test_base.DbTestCase): + def setUp(self): + super(TestSqliteUniqueConstraints, self).setUp() + + migration.patch_migrate() + + self.helper = sqlite.SQLiteHelper() + + sa.Table( + 'test_table', + sa.schema.MetaData(bind=session.get_engine()), + sa.Column('a', sa.Integer), + sa.Column('b', sa.String(10)), + sa.Column('c', sa.Integer), + sa.UniqueConstraint('a', 'b', name='unique_a_b'), + sa.UniqueConstraint('b', 'c', name='unique_b_c') + ).create() + + # NOTE(rpodolyaka): it's important to use the reflected table here + # rather than original one because this is what + # we actually do in db migrations code + self.reflected_table = sa.Table( + 'test_table', + sa.schema.MetaData(bind=session.get_engine()), + autoload=True + ) + + def test_get_unique_constraints(self): + table = self.reflected_table + + existing = uniques(*self.helper._get_unique_constraints(table)) + should_be = uniques( + sa.UniqueConstraint(table.c.a, table.c.b, name='unique_a_b'), + sa.UniqueConstraint(table.c.b, table.c.c, name='unique_b_c'), + ) + self.assertEquals(should_be, existing) + + def test_add_unique_constraint(self): + table = self.reflected_table + UniqueConstraint(table.c.a, table.c.c, name='unique_a_c').create() + + existing = uniques(*self.helper._get_unique_constraints(table)) + should_be = uniques( + sa.UniqueConstraint(table.c.a, table.c.b, name='unique_a_b'), + sa.UniqueConstraint(table.c.b, table.c.c, name='unique_b_c'), + sa.UniqueConstraint(table.c.a, table.c.c, name='unique_a_c'), + ) + self.assertEquals(should_be, existing) + + def test_drop_unique_constraint(self): + table = self.reflected_table + UniqueConstraint(table.c.a, table.c.b, name='unique_a_b').drop() + + existing = uniques(*self.helper._get_unique_constraints(table)) + should_be = uniques( + sa.UniqueConstraint(table.c.b, table.c.c, name='unique_b_c'), + ) + self.assertEquals(should_be, existing) -- cgit