diff options
Diffstat (limited to 'oslo_db/tests')
-rw-r--r-- | oslo_db/tests/fixtures.py | 11 | ||||
-rw-r--r-- | oslo_db/tests/sqlalchemy/test_exc_filters.py | 56 | ||||
-rw-r--r-- | oslo_db/tests/sqlalchemy/test_sqlalchemy.py | 7 | ||||
-rw-r--r-- | oslo_db/tests/sqlalchemy/test_utils.py | 29 |
4 files changed, 64 insertions, 39 deletions
diff --git a/oslo_db/tests/fixtures.py b/oslo_db/tests/fixtures.py index 521639c..502719b 100644 --- a/oslo_db/tests/fixtures.py +++ b/oslo_db/tests/fixtures.py @@ -49,11 +49,6 @@ class WarningsFixture(fixtures.Fixture): warnings.filterwarnings( 'once', - message=r'The current statement is being autocommitted .*', - category=sqla_exc.SADeprecationWarning) - - warnings.filterwarnings( - 'once', message=r'Calling \.begin\(\) when a transaction is already .*', category=sqla_exc.SADeprecationWarning) @@ -69,6 +64,12 @@ class WarningsFixture(fixtures.Fixture): category=sqla_exc.SADeprecationWarning) warnings.filterwarnings( + 'once', + message=r'The current statement is being autocommitted .*', + module='migrate', + category=sqla_exc.SADeprecationWarning) + + warnings.filterwarnings( 'ignore', message=r'The Engine.execute\(\) method is considered legacy .*', module='migrate', diff --git a/oslo_db/tests/sqlalchemy/test_exc_filters.py b/oslo_db/tests/sqlalchemy/test_exc_filters.py index 528d3ed..53789f5 100644 --- a/oslo_db/tests/sqlalchemy/test_exc_filters.py +++ b/oslo_db/tests/sqlalchemy/test_exc_filters.py @@ -498,9 +498,15 @@ class TestReferenceErrorSQLite( self.table_2.create(self.engine) def test_raise(self): - with self.engine.connect() as conn: - conn.execute(sql.text("PRAGMA foreign_keys = ON")) + connection = self.engine.raw_connection() + try: + cursor = connection.cursor() + cursor.execute('PRAGMA foreign_keys = ON') + cursor.close() + finally: + connection.close() + with self.engine.connect() as conn: matched = self.assertRaises( exception.DBReferenceError, conn.execute, @@ -521,16 +527,24 @@ class TestReferenceErrorSQLite( self.assertIsNone(matched.key_table) def test_raise_delete(self): - with self.engine.connect() as conn: - conn.execute(sql.text("PRAGMA foreign_keys = ON")) - conn.execute(self.table_1.insert().values(id=1234, foo=42)) - conn.execute(self.table_2.insert().values(id=4321, foo_id=1234)) + connection = self.engine.raw_connection() + try: + cursor = connection.cursor() + cursor.execute('PRAGMA foreign_keys = ON') + cursor.close() + finally: + connection.close() - matched = self.assertRaises( - exception.DBReferenceError, - conn.execute, - self.table_1.delete() - ) + with self.engine.connect() as conn: + with conn.begin(): + conn.execute(self.table_1.insert().values(id=1234, foo=42)) + conn.execute( + self.table_2.insert().values(id=4321, foo_id=1234)) + matched = self.assertRaises( + exception.DBReferenceError, + conn.execute, + self.table_1.delete() + ) self.assertInnerException( matched, @@ -577,13 +591,17 @@ class TestReferenceErrorPostgreSQL( def test_raise_delete(self): with self.engine.connect() as conn: - conn.execute(self.table_1.insert().values(id=1234, foo=42)) - conn.execute(self.table_2.insert().values(id=4321, foo_id=1234)) - matched = self.assertRaises( - exception.DBReferenceError, - conn.execute, - self.table_1.delete() - ) + with conn.begin(): + conn.execute(self.table_1.insert().values(id=1234, foo=42)) + conn.execute( + self.table_2.insert().values(id=4321, foo_id=1234)) + + with conn.begin(): + matched = self.assertRaises( + exception.DBReferenceError, + conn.execute, + self.table_1.delete() + ) self.assertInnerException( matched, @@ -648,7 +666,7 @@ class TestReferenceErrorMySQL( self.assertEqual("resource_foo", matched.key_table) def test_raise_delete(self): - with self.engine.connect() as conn: + with self.engine.connect() as conn, conn.begin(): conn.execute(self.table_1.insert().values(id=1234, foo=42)) conn.execute(self.table_2.insert().values(id=4321, foo_id=1234)) matched = self.assertRaises( diff --git a/oslo_db/tests/sqlalchemy/test_sqlalchemy.py b/oslo_db/tests/sqlalchemy/test_sqlalchemy.py index de2a6dc..7b634f1 100644 --- a/oslo_db/tests/sqlalchemy/test_sqlalchemy.py +++ b/oslo_db/tests/sqlalchemy/test_sqlalchemy.py @@ -314,12 +314,15 @@ class MySQLModeTestCase(db_test_base._MySQLOpportunisticTestCase): self.test_table = Table(_TABLE_NAME + "mode", meta, Column('id', Integer, primary_key=True), Column('bar', String(255))) - self.test_table.create(self.connection) + with self.connection.begin(): + self.test_table.create(self.connection) def cleanup(): - self.test_table.drop(self.connection) + with self.connection.begin(): + self.test_table.drop(self.connection) self.connection.close() mode_engine.dispose() + self.addCleanup(cleanup) def _test_string_too_long(self, value): diff --git a/oslo_db/tests/sqlalchemy/test_utils.py b/oslo_db/tests/sqlalchemy/test_utils.py index 2cbf47d..087f7ec 100644 --- a/oslo_db/tests/sqlalchemy/test_utils.py +++ b/oslo_db/tests/sqlalchemy/test_utils.py @@ -699,8 +699,9 @@ class TestMigrationUtils(db_test_base._DbTestCase): Column('updated_at', DateTime)) test_table.create(engine) - with engine.connect() as conn: - conn.execute(test_table.insert(), values) + with engine.connect() as conn, conn.begin(): + with conn.begin(): + conn.execute(test_table.insert(), values) return test_table, values def test_drop_old_duplicate_entries_from_table(self): @@ -720,7 +721,7 @@ class TestMigrationUtils(db_test_base._DbTestCase): uniq_values.add(uniq_value) expected_ids.append(value['id']) - with self.engine.connect() as conn: + with self.engine.connect() as conn, conn.begin(): real_ids = [ row[0] for row in conn.execute(select(test_table.c.id)).fetchall() @@ -762,7 +763,7 @@ class TestMigrationUtils(db_test_base._DbTestCase): base_select = table.select() - with self.engine.connect() as conn: + with self.engine.connect() as conn, conn.begin(): rows_select = base_select.where(table.c.deleted != table.c.id) row_ids = [ row.id for row in conn.execute(rows_select).fetchall() @@ -938,7 +939,7 @@ class TestMigrationUtils(db_test_base._DbTestCase): # NOTE(zzzeek): SQLAlchemy 1.2 Boolean type will disallow non 1/0 # value here, 1.1 also coerces to "1/0" so use raw SQL to test the # constraint - with self.engine.connect() as conn: + with self.engine.connect() as conn, conn.begin(): conn.exec_driver_sql( "INSERT INTO abc (deleted) VALUES (?)", (10, ), @@ -1652,7 +1653,7 @@ class TestDialectFunctionDispatcher(test_base.BaseTestCase): class TestGetInnoDBTables(db_test_base._MySQLOpportunisticTestCase): def test_all_tables_use_innodb(self): - with self.engine.connect() as conn: + with self.engine.connect() as conn, conn.begin(): conn.execute( sql.text( "CREATE TABLE customers " @@ -1660,21 +1661,23 @@ class TestGetInnoDBTables(db_test_base._MySQLOpportunisticTestCase): self.assertEqual([], utils.get_non_innodb_tables(self.engine)) def test_all_tables_use_innodb_false(self): - with self.engine.connect() as conn: + with self.engine.connect() as conn, conn.begin(): conn.execute( - sql.text("CREATE TABLE employee (i INT) ENGINE=MEMORY")) + sql.text("CREATE TABLE employee (i INT) ENGINE=MEMORY") + ) self.assertEqual(['employee'], utils.get_non_innodb_tables(self.engine)) def test_skip_tables_use_default_value(self): - with self.engine.connect() as conn: + with self.engine.connect() as conn, conn.begin(): conn.execute( - sql.text("CREATE TABLE migrate_version (i INT) ENGINE=MEMORY")) + sql.text("CREATE TABLE migrate_version (i INT) ENGINE=MEMORY") + ) self.assertEqual([], utils.get_non_innodb_tables(self.engine)) def test_skip_tables_use_passed_value(self): - with self.engine.connect() as conn: + with self.engine.connect() as conn, conn.begin(): conn.execute( sql.text("CREATE TABLE some_table (i INT) ENGINE=MEMORY")) self.assertEqual([], @@ -1682,7 +1685,7 @@ class TestGetInnoDBTables(db_test_base._MySQLOpportunisticTestCase): self.engine, skip_tables=('some_table',))) def test_skip_tables_use_empty_list(self): - with self.engine.connect() as conn: + with self.engine.connect() as conn, conn.begin(): conn.execute( sql.text("CREATE TABLE some_table_3 (i INT) ENGINE=MEMORY")) self.assertEqual(['some_table_3'], @@ -1690,7 +1693,7 @@ class TestGetInnoDBTables(db_test_base._MySQLOpportunisticTestCase): self.engine, skip_tables=())) def test_skip_tables_use_several_values(self): - with self.engine.connect() as conn: + with self.engine.connect() as conn, conn.begin(): conn.execute( sql.text("CREATE TABLE some_table_1 (i INT) ENGINE=MEMORY")) conn.execute( |