summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2014-10-07 17:42:26 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2014-10-08 14:47:56 -0400
commit33011a51b23af06eee29a493b6c7efacc7be0b1e (patch)
tree979d34ea1b55b9cf3e78554a74ea4018a53f1517
parent01a54cc5cc43c8b497007dbb07c78dacd03c1ed2 (diff)
downloadoslo-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.py43
-rw-r--r--tests/sqlalchemy/test_utils.py128
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__"