diff options
author | Mike Bayer <mike_mp@zzzcomputing.com> | 2014-10-07 17:42:26 -0400 |
---|---|---|
committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2014-10-08 14:47:56 -0400 |
commit | 33011a51b23af06eee29a493b6c7efacc7be0b1e (patch) | |
tree | 979d34ea1b55b9cf3e78554a74ea4018a53f1517 | |
parent | 01a54cc5cc43c8b497007dbb07c78dacd03c1ed2 (diff) | |
download | oslo-db-33011a51b23af06eee29a493b6c7efacc7be0b1e.tar.gz |
Remove utils.drop_unique_constraint()
This function is currently otherwise unused and does not
work correctly on SQLite, not only dropping the
target unique constraint but instead dropping all unique
constraints from the target table prior to SQLAlchemy
version 1.0 (unreleased). The tests miss that it isn't
working as well unless SQLAlchemy 1.0 is used, in which
case they fail.
Change-Id: Icaff621fba4df289247e96b87c439c62e91543d6
Closes-bug: #1377646
-rw-r--r-- | oslo/db/sqlalchemy/utils.py | 43 | ||||
-rw-r--r-- | tests/sqlalchemy/test_utils.py | 128 |
2 files changed, 1 insertions, 170 deletions
diff --git a/oslo/db/sqlalchemy/utils.py b/oslo/db/sqlalchemy/utils.py index 71cd1ea..e2537a6 100644 --- a/oslo/db/sqlalchemy/utils.py +++ b/oslo/db/sqlalchemy/utils.py @@ -375,49 +375,6 @@ def _get_not_supported_column(col_name_col_instance, column_name): return column -def drop_unique_constraint(migrate_engine, table_name, uc_name, *columns, - **col_name_col_instance): - """Drop unique constraint from table. - - DEPRECATED: this function is deprecated and will be removed from oslo.db - in a few releases. Please use UniqueConstraint.drop() method directly for - sqlalchemy-migrate migration scripts. - - This method drops UC from table and works for mysql, postgresql and sqlite. - In mysql and postgresql we are able to use "alter table" construction. - Sqlalchemy doesn't support some sqlite column types and replaces their - type with NullType in metadata. We process these columns and replace - NullType with the correct column type. - - :param migrate_engine: sqlalchemy engine - :param table_name: name of table that contains uniq constraint. - :param uc_name: name of uniq constraint that will be dropped. - :param columns: columns that are in uniq constraint. - :param col_name_col_instance: contains pair column_name=column_instance. - column_instance is instance of Column. These params - are required only for columns that have unsupported - types by sqlite. For example BigInteger. - """ - - from migrate.changeset import UniqueConstraint - - meta = MetaData() - meta.bind = migrate_engine - t = Table(table_name, meta, autoload=True) - - if migrate_engine.name == "sqlite": - override_cols = [ - _get_not_supported_column(col_name_col_instance, col.name) - for col in t.columns - if isinstance(col.type, NullType) - ] - for col in override_cols: - t.columns.replace(col) - - uc = UniqueConstraint(*columns, table=t, name=uc_name) - uc.drop() - - def drop_old_duplicate_entries_from_table(migrate_engine, table_name, use_soft_delete, *uc_column_names): """Drop all old rows having the same values for columns in uc_columns. diff --git a/tests/sqlalchemy/test_utils.py b/tests/sqlalchemy/test_utils.py index 742cd38..acf31d0 100644 --- a/tests/sqlalchemy/test_utils.py +++ b/tests/sqlalchemy/test_utils.py @@ -14,22 +14,19 @@ # under the License. import uuid -import warnings import fixtures -from migrate.changeset import UniqueConstraint import mock from oslotest import base as test_base from oslotest import moxstubout import six -from six import moves from six.moves.urllib import parse import sqlalchemy from sqlalchemy.dialects import mysql from sqlalchemy import Boolean, Index, Integer, DateTime, String, SmallInteger from sqlalchemy import MetaData, Table, Column, ForeignKey from sqlalchemy.engine import reflection -from sqlalchemy.exc import SAWarning, ResourceClosedError +from sqlalchemy.exc import ResourceClosedError from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.sql import select from sqlalchemy.types import UserDefinedType, NullType @@ -453,129 +450,6 @@ class TestMigrationUtils(db_test_base.DbTestCase): # behavior), any integer value can be inserted, otherwise only 1 or 0. self.engine.execute(table.insert({'deleted': 10})) - def test_utils_drop_unique_constraint(self): - table_name = "__test_tmp_table__" - uc_name = 'uniq_foo' - values = [ - {'id': 1, 'a': 3, 'foo': 10}, - {'id': 2, 'a': 2, 'foo': 20}, - {'id': 3, 'a': 1, 'foo': 30}, - ] - test_table = Table( - table_name, self.meta, - Column('id', Integer, primary_key=True, nullable=False), - Column('a', Integer), - Column('foo', Integer), - UniqueConstraint('a', name='uniq_a'), - UniqueConstraint('foo', name=uc_name), - ) - test_table.create() - - self.engine.execute(test_table.insert(), values) - # NOTE(boris-42): This method is generic UC dropper. - utils.drop_unique_constraint(self.engine, table_name, uc_name, 'foo') - - s = test_table.select().order_by(test_table.c.id) - rows = self.engine.execute(s).fetchall() - - for i in moves.range(len(values)): - v = values[i] - self.assertEqual((v['id'], v['a'], v['foo']), rows[i]) - - # NOTE(boris-42): Update data about Table from DB. - meta = MetaData(bind=self.engine) - test_table = Table(table_name, meta, autoload=True) - constraints = [c for c in test_table.constraints - if c.name == uc_name] - self.assertEqual(len(constraints), 0) - self.assertEqual(len(test_table.constraints), 1) - - test_table.drop() - - @db_test_base.backend_specific('sqlite') - def test_util_drop_unique_constraint_with_not_supported_sqlite_type(self): - table_name = "__test_tmp_table__" - uc_name = 'uniq_foo' - values = [ - {'id': 1, 'a': 3, 'foo': 10}, - {'id': 2, 'a': 2, 'foo': 20}, - {'id': 3, 'a': 1, 'foo': 30} - ] - - test_table = Table( - table_name, self.meta, - Column('id', Integer, primary_key=True, nullable=False), - Column('a', Integer), - Column('foo', CustomType, default=0), - UniqueConstraint('a', name='uniq_a'), - UniqueConstraint('foo', name=uc_name), - ) - test_table.create() - - self.engine.execute(test_table.insert(), values) - warnings.simplefilter("ignore", SAWarning) - - # reflection of custom types has been fixed upstream - if SA_VERSION < (0, 9, 0): - # NOTE(boris-42): Missing info about column `foo` that has - # unsupported type CustomType. - self.assertRaises(exception.ColumnError, - utils.drop_unique_constraint, - self.engine, table_name, uc_name, 'foo') - - # NOTE(boris-42): Wrong type of foo instance. it should be - # instance of sqlalchemy.Column. - self.assertRaises(exception.ColumnError, - utils.drop_unique_constraint, - self.engine, table_name, uc_name, 'foo', - foo=Integer()) - - foo = Column('foo', CustomType, default=0) - utils.drop_unique_constraint( - self.engine, table_name, uc_name, 'foo', foo=foo) - - s = test_table.select().order_by(test_table.c.id) - rows = self.engine.execute(s).fetchall() - - for i in moves.range(len(values)): - v = values[i] - self.assertEqual((v['id'], v['a'], v['foo']), rows[i]) - - # NOTE(boris-42): Update data about Table from DB. - meta = MetaData(bind=self.engine) - test_table = Table(table_name, meta, autoload=True) - constraints = [c for c in test_table.constraints if c.name == uc_name] - self.assertEqual(len(constraints), 0) - self.assertEqual(len(test_table.constraints), 1) - test_table.drop() - - @db_test_base.backend_specific('sqlite') - def test_drop_unique_constraint_in_sqlite_fk_recreate(self): - parent_table = Table( - 'table0', self.meta, - Column('id', Integer, primary_key=True), - Column('foo', Integer), - ) - parent_table.create() - table_name = 'table1' - table = Table( - table_name, self.meta, - Column('id', Integer, primary_key=True), - Column('baz', Integer), - Column('bar', Integer, ForeignKey("table0.id")), - UniqueConstraint('baz', name='constr1') - ) - table.create() - utils.drop_unique_constraint(self.engine, table_name, 'constr1', 'baz') - - insp = reflection.Inspector.from_engine(self.engine) - f_keys = insp.get_foreign_keys(table_name) - self.assertEqual(len(f_keys), 1) - f_key = f_keys[0] - self.assertEqual(f_key['referred_table'], 'table0') - self.assertEqual(f_key['referred_columns'], ['id']) - self.assertEqual(f_key['constrained_columns'], ['bar']) - def test_insert_from_select(self): insert_table_name = "__test_insert_to_table__" select_table_name = "__test_select_from_table__" |