diff options
Diffstat (limited to 'oslo_db/tests/sqlalchemy/test_provision.py')
-rw-r--r-- | oslo_db/tests/sqlalchemy/test_provision.py | 76 |
1 files changed, 0 insertions, 76 deletions
diff --git a/oslo_db/tests/sqlalchemy/test_provision.py b/oslo_db/tests/sqlalchemy/test_provision.py index f0ef944..a6cedce 100644 --- a/oslo_db/tests/sqlalchemy/test_provision.py +++ b/oslo_db/tests/sqlalchemy/test_provision.py @@ -22,7 +22,6 @@ from oslo_db import exception from oslo_db.sqlalchemy import enginefacade from oslo_db.sqlalchemy import provision from oslo_db.sqlalchemy import test_fixtures -from oslo_db.sqlalchemy import utils from oslo_db.tests import base as test_base from oslo_db.tests.sqlalchemy import base as db_test_base @@ -149,81 +148,6 @@ class PostgreSQLDropAllObjectsTest( pass -class RetainSchemaTest(test_base.BaseTestCase): - DRIVER = "sqlite" - - def setUp(self): - super(RetainSchemaTest, self).setUp() - - metadata = schema.MetaData() - self.test_table = schema.Table( - 'test_table', metadata, - schema.Column('x', types.Integer), - schema.Column('y', types.Integer), - mysql_engine='InnoDB' - ) - - def gen_schema(engine): - metadata.create_all(engine, checkfirst=False) - self._gen_schema = gen_schema - - def test_once(self): - self._run_test() - - def test_twice(self): - self._run_test() - - def _run_test(self): - try: - database_resource = provision.DatabaseResource( - self.DRIVER, provision_new_database=True) - except exception.BackendNotAvailable: - self.skipTest("database not available") - - schema_resource = provision.SchemaResource( - database_resource, self._gen_schema) - - schema = schema_resource.getResource() - - conn = schema.database.engine.connect() - engine = utils.NonCommittingEngine(conn) - - with engine.connect() as conn: - rows = conn.execute(self.test_table.select()) - self.assertEqual([], rows.fetchall()) - - trans = conn.begin() - conn.execute( - self.test_table.insert(), - {"x": 1, "y": 2} - ) - trans.rollback() - - rows = conn.execute(self.test_table.select()) - self.assertEqual([], rows.fetchall()) - - trans = conn.begin() - conn.execute( - self.test_table.insert(), - {"x": 2, "y": 3} - ) - trans.commit() - - rows = conn.execute(self.test_table.select()) - self.assertEqual([(2, 3)], rows.fetchall()) - - engine._dispose() - schema_resource.finishedWith(schema) - - -class MySQLRetainSchemaTest(RetainSchemaTest): - DRIVER = "mysql" - - -class PostgresqlRetainSchemaTest(RetainSchemaTest): - DRIVER = "postgresql" - - class AdHocURLTest(test_base.BaseTestCase): def test_sqlite_setup_teardown(self): |