diff options
Diffstat (limited to 'oslo_db/tests')
-rw-r--r-- | oslo_db/tests/sqlalchemy/__init__.py | 18 | ||||
-rw-r--r-- | oslo_db/tests/sqlalchemy/test_exc_filters.py | 17 | ||||
-rw-r--r-- | oslo_db/tests/sqlalchemy/test_migrations.py | 2 | ||||
-rw-r--r-- | oslo_db/tests/sqlalchemy/test_provision.py | 82 | ||||
-rw-r--r-- | oslo_db/tests/sqlalchemy/test_sqlalchemy.py | 18 |
5 files changed, 118 insertions, 19 deletions
diff --git a/oslo_db/tests/sqlalchemy/__init__.py b/oslo_db/tests/sqlalchemy/__init__.py index e69de29..e4bfd87 100644 --- a/oslo_db/tests/sqlalchemy/__init__.py +++ b/oslo_db/tests/sqlalchemy/__init__.py @@ -0,0 +1,18 @@ +# Copyright (c) 2014 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 oslo.db.sqlalchemy import test_base + +load_tests = test_base.optimize_db_test_loader(__file__) diff --git a/oslo_db/tests/sqlalchemy/test_exc_filters.py b/oslo_db/tests/sqlalchemy/test_exc_filters.py index edab9d6..aafdcfb 100644 --- a/oslo_db/tests/sqlalchemy/test_exc_filters.py +++ b/oslo_db/tests/sqlalchemy/test_exc_filters.py @@ -326,12 +326,17 @@ class TestReferenceErrorMySQL(TestReferenceErrorSQLite, self.assertEqual("resource_foo", matched.key_table) def test_raise_ansi_quotes(self): - self.engine.execute("SET SESSION sql_mode = 'ANSI';") - matched = self.assertRaises( - exception.DBReferenceError, - self.engine.execute, - self.table_2.insert({'id': 1, 'foo_id': 2}) - ) + with self.engine.connect() as conn: + conn.detach() # will not be returned to the pool when closed + + # this is incompatible with some internals of the engine + conn.execute("SET SESSION sql_mode = 'ANSI';") + + matched = self.assertRaises( + exception.DBReferenceError, + conn.execute, + self.table_2.insert({'id': 1, 'foo_id': 2}) + ) self.assertInnerException( matched, diff --git a/oslo_db/tests/sqlalchemy/test_migrations.py b/oslo_db/tests/sqlalchemy/test_migrations.py index 1bae701..96e7e31 100644 --- a/oslo_db/tests/sqlalchemy/test_migrations.py +++ b/oslo_db/tests/sqlalchemy/test_migrations.py @@ -171,7 +171,7 @@ class TestWalkVersions(test.BaseTestCase, migrate.WalkVersionsMixin): self.assertEqual(upgraded, self.migrate_up.call_args_list) -class ModelsMigrationSyncMixin(test.BaseTestCase): +class ModelsMigrationSyncMixin(test_base.DbTestCase): def setUp(self): super(ModelsMigrationSyncMixin, self).setUp() diff --git a/oslo_db/tests/sqlalchemy/test_provision.py b/oslo_db/tests/sqlalchemy/test_provision.py index 7c57de3..25f5bc4 100644 --- a/oslo_db/tests/sqlalchemy/test_provision.py +++ b/oslo_db/tests/sqlalchemy/test_provision.py @@ -10,11 +10,13 @@ # License for the specific language governing permissions and limitations # under the License. - +from oslotest import base as oslo_test_base from sqlalchemy import inspect from sqlalchemy import schema from sqlalchemy import types +from oslo.db import exception +from oslo.db.sqlalchemy import provision from oslo_db.sqlalchemy import test_base @@ -62,7 +64,7 @@ class DropAllObjectsTest(test_base.DbTestCase): set(insp.get_table_names()) ) - self.provision.drop_all_objects() + self.db.backend.drop_all_objects(self.engine) insp = inspect(self.engine) self.assertEqual( @@ -71,11 +73,83 @@ class DropAllObjectsTest(test_base.DbTestCase): ) -class MySQLRetainSchemaTest( +class MySQLDropAllObjectsTest( DropAllObjectsTest, test_base.MySQLOpportunisticTestCase): pass -class PostgresqlRetainSchemaTest( +class PostgreSQLDropAllObjectsTest( DropAllObjectsTest, test_base.PostgreSQLOpportunisticTestCase): pass + + +class RetainSchemaTest(oslo_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) + except exception.BackendNotAvailable: + self.skip("database not available") + + schema_resource = provision.SchemaResource( + database_resource, self._gen_schema) + transaction_resource = provision.TransactionResource( + database_resource, schema_resource) + + engine = transaction_resource.getResource() + + 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(rows.fetchall(), [(2, 3)]) + + transaction_resource.finishedWith(engine) + + +class MySQLRetainSchemaTest(RetainSchemaTest): + DRIVER = "mysql" + + +class PostgresqlRetainSchemaTest(RetainSchemaTest): + DRIVER = "postgresql" diff --git a/oslo_db/tests/sqlalchemy/test_sqlalchemy.py b/oslo_db/tests/sqlalchemy/test_sqlalchemy.py index bcc4a1c..24aeb22 100644 --- a/oslo_db/tests/sqlalchemy/test_sqlalchemy.py +++ b/oslo_db/tests/sqlalchemy/test_sqlalchemy.py @@ -229,20 +229,22 @@ class MySQLModeTestCase(test_base.MySQLOpportunisticTestCase): def setUp(self): super(MySQLModeTestCase, self).setUp() - - self.engine = session.create_engine(self.engine.url, - mysql_sql_mode=self.mysql_mode) - self.connection = self.engine.connect() + mode_engine = session.create_engine( + self.engine.url, + mysql_sql_mode=self.mysql_mode) + self.connection = mode_engine.connect() meta = MetaData() - meta.bind = self.engine self.test_table = Table(_TABLE_NAME + "mode", meta, Column('id', Integer, primary_key=True), Column('bar', String(255))) - self.test_table.create() + self.test_table.create(self.connection) - self.addCleanup(self.test_table.drop) - self.addCleanup(self.connection.close) + def cleanup(): + self.test_table.drop(self.connection) + self.connection.close() + mode_engine.dispose() + self.addCleanup(cleanup) def _test_string_too_long(self, value): with self.connection.begin(): |