2ad571c8d7
The original idea of enginefacade was that applications would call upon the global _TransactionContextManager given in oslo_db.sqlalchemy.enginefacade. However, as it turns out, virtually no Openstack projects seem to be using that technique, and instead, everyone is creating their own ad-hoc _TransactionContextManager objects and usually establishing it as a module-level global. Nova has two of them. Additionally, projects add configuration to these enginefacades (which IS part of the original idea), and this configuration in some cases is necessary to be present for tests that run as well, a key example being the sqlite_fks flag. The original DbFixture integration provided no way of reusing this configuration. Finally, projects very much tend to use custom fixtures in order to define their database communication. Test classes themselves don't really make use of oslo_db's DbTestCase anymore. This patch introduces a modernized fixture system which, in conjunction with the recent provisioning patch, addresses these use cases. Applications will typically create their own subclasses of these fixtures up front to suit the various testing cases they have, including SQLite fixed, SQLite ad-hoc, and opportunistic. In order to accommodate the fixture-based flow along with the use of testresources for opportunistic database provisioning, a mixin class OpportunisticDbTestMixin is still needed when a test needs to use "opportunistic" testing in order to provide the .resources attribute. The calculation of .resources is moved into the fixture system, but because this attribute is consulted before setUp(), the "opportunistic" fixture must be created early and stored. Closes-Bug: #1548960 Change-Id: I0163e637ffef6d45d2573ebe29b5438911d01fce
265 lines
8.2 KiB
Python
265 lines
8.2 KiB
Python
# 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.
|
|
|
|
import mock
|
|
import os
|
|
|
|
from oslotest import base as oslo_test_base
|
|
from sqlalchemy import exc as sa_exc
|
|
from sqlalchemy import inspect
|
|
from sqlalchemy import schema
|
|
from sqlalchemy import types
|
|
|
|
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.sqlalchemy import base as test_base
|
|
|
|
|
|
class DropAllObjectsTest(test_base.DbTestCase):
|
|
|
|
def setUp(self):
|
|
super(DropAllObjectsTest, self).setUp()
|
|
|
|
self.metadata = metadata = schema.MetaData()
|
|
schema.Table(
|
|
'a', metadata,
|
|
schema.Column('id', types.Integer, primary_key=True),
|
|
mysql_engine='InnoDB'
|
|
)
|
|
schema.Table(
|
|
'b', metadata,
|
|
schema.Column('id', types.Integer, primary_key=True),
|
|
schema.Column('a_id', types.Integer, schema.ForeignKey('a.id')),
|
|
mysql_engine='InnoDB'
|
|
)
|
|
schema.Table(
|
|
'c', metadata,
|
|
schema.Column('id', types.Integer, primary_key=True),
|
|
schema.Column('b_id', types.Integer, schema.ForeignKey('b.id')),
|
|
schema.Column(
|
|
'd_id', types.Integer,
|
|
schema.ForeignKey('d.id', use_alter=True, name='c_d_fk')),
|
|
mysql_engine='InnoDB'
|
|
)
|
|
schema.Table(
|
|
'd', metadata,
|
|
schema.Column('id', types.Integer, primary_key=True),
|
|
schema.Column('c_id', types.Integer, schema.ForeignKey('c.id')),
|
|
mysql_engine='InnoDB'
|
|
)
|
|
|
|
metadata.create_all(self.engine, checkfirst=False)
|
|
# will drop nothing if the test worked
|
|
self.addCleanup(metadata.drop_all, self.engine, checkfirst=True)
|
|
|
|
def test_drop_all(self):
|
|
insp = inspect(self.engine)
|
|
self.assertEqual(
|
|
set(['a', 'b', 'c', 'd']),
|
|
set(insp.get_table_names())
|
|
)
|
|
|
|
self._get_default_provisioned_db().\
|
|
backend.drop_all_objects(self.engine)
|
|
|
|
insp = inspect(self.engine)
|
|
self.assertEqual(
|
|
[],
|
|
insp.get_table_names()
|
|
)
|
|
|
|
|
|
class BackendNotAvailableTest(oslo_test_base.BaseTestCase):
|
|
def test_no_dbapi(self):
|
|
backend = provision.Backend(
|
|
"postgresql", "postgresql+nosuchdbapi://hostname/dsn")
|
|
|
|
with mock.patch(
|
|
"sqlalchemy.create_engine",
|
|
mock.Mock(side_effect=ImportError("nosuchdbapi"))):
|
|
|
|
# NOTE(zzzeek): Call and test the _verify function twice, as it
|
|
# exercises a different code path on subsequent runs vs.
|
|
# the first run
|
|
ex = self.assertRaises(
|
|
exception.BackendNotAvailable,
|
|
backend._verify)
|
|
self.assertEqual(
|
|
"Backend 'postgresql+nosuchdbapi' is unavailable: "
|
|
"No DBAPI installed", str(ex))
|
|
|
|
ex = self.assertRaises(
|
|
exception.BackendNotAvailable,
|
|
backend._verify)
|
|
self.assertEqual(
|
|
"Backend 'postgresql+nosuchdbapi' is unavailable: "
|
|
"No DBAPI installed", str(ex))
|
|
|
|
def test_cant_connect(self):
|
|
backend = provision.Backend(
|
|
"postgresql", "postgresql+nosuchdbapi://hostname/dsn")
|
|
|
|
with mock.patch(
|
|
"sqlalchemy.create_engine",
|
|
mock.Mock(return_value=mock.Mock(connect=mock.Mock(
|
|
side_effect=sa_exc.OperationalError(
|
|
"can't connect", None, None))
|
|
))
|
|
):
|
|
|
|
# NOTE(zzzeek): Call and test the _verify function twice, as it
|
|
# exercises a different code path on subsequent runs vs.
|
|
# the first run
|
|
ex = self.assertRaises(
|
|
exception.BackendNotAvailable,
|
|
backend._verify)
|
|
self.assertEqual(
|
|
"Backend 'postgresql+nosuchdbapi' is unavailable: "
|
|
"Could not connect", str(ex))
|
|
|
|
ex = self.assertRaises(
|
|
exception.BackendNotAvailable,
|
|
backend._verify)
|
|
self.assertEqual(
|
|
"Backend 'postgresql+nosuchdbapi' is unavailable: "
|
|
"Could not connect", str(ex))
|
|
|
|
|
|
class MySQLDropAllObjectsTest(
|
|
DropAllObjectsTest, test_base.MySQLOpportunisticTestCase):
|
|
pass
|
|
|
|
|
|
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, provision_new_database=True)
|
|
except exception.BackendNotAvailable:
|
|
self.skip("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(oslo_test_base.BaseTestCase):
|
|
def test_sqlite_setup_teardown(self):
|
|
|
|
fixture = test_fixtures.AdHocDbFixture("sqlite:///foo.db")
|
|
|
|
fixture.setUp()
|
|
|
|
self.assertEqual(
|
|
str(enginefacade._context_manager._factory._writer_engine.url),
|
|
"sqlite:///foo.db"
|
|
)
|
|
|
|
self.assertTrue(os.path.exists("foo.db"))
|
|
fixture.cleanUp()
|
|
|
|
self.assertFalse(os.path.exists("foo.db"))
|
|
|
|
def test_mysql_setup_teardown(self):
|
|
try:
|
|
mysql_backend = provision.Backend.backend_for_database_type(
|
|
"mysql")
|
|
except exception.BackendNotAvailable:
|
|
self.skip("mysql backend not available")
|
|
|
|
mysql_backend.create_named_database("adhoc_test")
|
|
self.addCleanup(
|
|
mysql_backend.drop_named_database, "adhoc_test"
|
|
)
|
|
url = str(mysql_backend.provisioned_database_url("adhoc_test"))
|
|
|
|
fixture = test_fixtures.AdHocDbFixture(url)
|
|
|
|
fixture.setUp()
|
|
|
|
self.assertEqual(
|
|
str(enginefacade._context_manager._factory._writer_engine.url),
|
|
url
|
|
)
|
|
|
|
fixture.cleanUp()
|