From 957b7f881280b31bfa122166abf20de0c01d6df9 Mon Sep 17 00:00:00 2001 From: Rick Harris Date: Thu, 21 Feb 2013 22:02:03 +0000 Subject: [PATCH] Fix baremetal migration skipping. The `have_mysql` function would only check for 'openstack_ci_test' and not 'openstack_baremetal_test' yet was being used for the baremetal-migration tests. This resulted in tests to run that should have been skipped. The solution is to have the test classes to define their configuraiton requirements and then to pass these into the backend checking code explicitly. Change-Id: I436b5cfe4cace04e7bcccc2bed75b34948fe0272 (cherry picked from commit 1c5518badd5a74fecb7cf8b89d37926a5eeac7ab) --- nova/tests/test_migrations.py | 145 ++++++++++++++++------------------ 1 file changed, 66 insertions(+), 79 deletions(-) diff --git a/nova/tests/test_migrations.py b/nova/tests/test_migrations.py index f800d2eb7a4f..2e5a01a6a322 100644 --- a/nova/tests/test_migrations.py +++ b/nova/tests/test_migrations.py @@ -63,41 +63,25 @@ import nova.virt.baremetal.db.sqlalchemy.migrate_repo LOG = logging.getLogger(__name__) -def _get_connect_string(backend, - user=None, - passwd=None, - database=None): +def _get_connect_string(backend, user, passwd, database): """ Try to get a connection with a very specific set of values, if we get these then we'll run the tests, otherwise they are skipped """ - if not user: - user = "openstack_citest" - if not passwd: - passwd = "openstack_citest" - if not database: - database = "openstack_citest" - if backend == "postgres": backend = "postgresql+psycopg2" elif backend == "mysql": backend = "mysql+mysqldb" + else: + raise Exception("Unrecognized backend: '%s'" % backend) return ("%(backend)s://%(user)s:%(passwd)s@localhost/%(database)s" % locals()) -def _is_backend_avail(backend, - user="openstack_citest", - passwd="openstack_citest", - database="openstack_citest"): +def _is_backend_avail(backend, user, passwd, database): try: - if backend == "mysql": - connect_uri = _get_connect_string("mysql", - user=user, passwd=passwd, database=database) - elif backend == "postgres": - connect_uri = _get_connect_string("postgres", - user=user, passwd=passwd, database=database) + connect_uri = _get_connect_string(backend, user, passwd, database) engine = sqlalchemy.create_engine(connect_uri) connection = engine.connect() except Exception: @@ -110,17 +94,17 @@ def _is_backend_avail(backend, return True -def _have_mysql(): +def _have_mysql(user, passwd, database): present = os.environ.get('NOVA_TEST_MYSQL_PRESENT') if present is None: - return _is_backend_avail('mysql') + return _is_backend_avail('mysql', user, passwd, database) return present.lower() in ('', 'true') -def _have_postgresql(): +def _have_postgresql(user, passwd, database): present = os.environ.get('NOVA_TEST_POSTGRESQL_PRESENT') if present is None: - return _is_backend_avail('postgres') + return _is_backend_avail('postgres', user, passwd, database) return present.lower() in ('', 'true') @@ -162,8 +146,49 @@ def get_pgsql_connection_info(conn_pieces): return (user, password, database, host) +class CommonTestsMixIn(object): + """These tests are shared between TestNovaMigrations and + TestBaremetalMigrations. + + BaseMigrationTestCase is effectively an abstract class, meant to be derived + from and not directly tested against; that's why these `test_` methods need + to be on a Mixin, so that they won't be picked up as valid tests for + BaseMigrationTestCase. + """ + def test_walk_versions(self): + for key, engine in self.engines.items(): + self._walk_versions(engine, self.snake_walk) + + def test_mysql_opportunistically(self): + self._test_mysql_opportunistically() + + def test_mysql_connect_fail(self): + """ + Test that we can trigger a mysql connection failure and we fail + gracefully to ensure we don't break people without mysql + """ + if _is_backend_avail('mysql', "openstack_cifail", self.PASSWD, + self.DATABASE): + self.fail("Shouldn't have connected") + + def test_postgresql_opportunistically(self): + self._test_postgresql_opportunistically() + + def test_postgresql_connect_fail(self): + """ + Test that we can trigger a postgres connection failure and we fail + gracefully to ensure we don't break people without postgres + """ + if _is_backend_avail('postgres', "openstack_cifail", self.PASSWD, + self.DATABASE): + self.fail("Shouldn't have connected") + + class BaseMigrationTestCase(test.TestCase): """Base class fort testing migrations and migration utils.""" + USER = None + PASSWD = None + DATABASE = None def __init__(self, *args, **kwargs): super(BaseMigrationTestCase, self).__init__(*args, **kwargs) @@ -270,21 +295,14 @@ class BaseMigrationTestCase(test.TestCase): os.unsetenv('PGPASSWORD') os.unsetenv('PGUSER') - def test_mysql_connect_fail(self): - """ - Test that we can trigger a mysql connection failure and we fail - gracefully to ensure we don't break people without mysql - """ - if _is_backend_avail('mysql', user="openstack_cifail"): - self.fail("Shouldn't have connected") - - def _test_mysql_opportunistically(self, database=None): + def _test_mysql_opportunistically(self): # Test that table creation on mysql only builds InnoDB tables - if not _have_mysql(): + if not _have_mysql(self.USER, self.PASSWD, self.DATABASE): self.skipTest("mysql not available") # add this to the global lists to make reset work with it, it's removed # automatically in tearDown so no need to clean it up here. - connect_string = _get_connect_string("mysql", database=database) + connect_string = _get_connect_string("mysql", self.USER, self.PASSWD, + self.DATABASE) (user, password, database, host) = \ get_mysql_connection_info(urlparse.urlparse(connect_string)) engine = sqlalchemy.create_engine(connect_string) @@ -313,21 +331,14 @@ class BaseMigrationTestCase(test.TestCase): self.assertEqual(count, 0, "%d non InnoDB tables created" % count) connection.close() - def test_postgresql_connect_fail(self): - """ - Test that we can trigger a postgres connection failure and we fail - gracefully to ensure we don't break people without postgres - """ - if _is_backend_avail('postgresql', user="openstack_cifail"): - self.fail("Shouldn't have connected") - - def _test_postgresql_opportunistically(self, database=None): + def _test_postgresql_opportunistically(self): # Test postgresql database migration walk - if not _have_postgresql(): + if not _have_postgresql(self.USER, self.PASSWD, self.DATABASE): self.skipTest("postgresql not available") # add this to the global lists to make reset work with it, it's removed # automatically in tearDown so no need to clean it up here. - connect_string = _get_connect_string("postgres", database=database) + connect_string = _get_connect_string("postgres", self.USER, + self.PASSWD, self.DATABASE) engine = sqlalchemy.create_engine(connect_string) (user, password, database, host) = \ get_mysql_connection_info(urlparse.urlparse(connect_string)) @@ -420,8 +431,11 @@ class BaseMigrationTestCase(test.TestCase): raise -class TestNovaMigrations(BaseMigrationTestCase): +class TestNovaMigrations(BaseMigrationTestCase, CommonTestsMixIn): """Test sqlalchemy-migrate migrations.""" + USER = "openstack_citest" + PASSWD = "openstack_citest" + DATABASE = "openstack_citest" def __init__(self, *args, **kwargs): super(TestNovaMigrations, self).__init__(*args, **kwargs) @@ -448,21 +462,6 @@ class TestNovaMigrations(BaseMigrationTestCase): globals(), locals(), ['versioning_api'], -1) self.migration_api = temp.versioning_api - def tearDown(self): - super(TestNovaMigrations, self).tearDown() - - def test_walk_versions(self): - for key, engine in self.engines.items(): - self._walk_versions(engine, self.snake_walk) - - def test_mysql_opportunistically(self): - self._test_mysql_opportunistically( - database='openstack_citest') - - def test_postgresql_opportunistically(self): - self._test_postgresql_opportunistically( - database='openstack_citest') - def _prerun_134(self, engine): now = timeutils.utcnow() data = [{ @@ -888,8 +887,11 @@ class TestNovaMigrations(BaseMigrationTestCase): self.assertEqual(len(rows), 1) -class TestBaremetalMigrations(BaseMigrationTestCase): +class TestBaremetalMigrations(BaseMigrationTestCase, CommonTestsMixIn): """Test sqlalchemy-migrate migrations.""" + USER = "openstack_baremetal_citest" + PASSWD = "openstack_baremetal_citest" + DATABASE = "openstack_baremetal_citest" def __init__(self, *args, **kwargs): super(TestBaremetalMigrations, self).__init__(*args, **kwargs) @@ -918,21 +920,6 @@ class TestBaremetalMigrations(BaseMigrationTestCase): globals(), locals(), ['versioning_api'], -1) self.migration_api = temp.versioning_api - def tearDown(self): - super(TestBaremetalMigrations, self).tearDown() - - def test_walk_versions(self): - for key, engine in self.engines.items(): - self._walk_versions(engine, self.snake_walk) - - def test_mysql_opportunistically(self): - self._test_mysql_opportunistically( - database='openstack_baremetal_citest') - - def test_postgresql_opportunistically(self): - self._test_postgresql_opportunistically( - database='openstack_baremetal_citest') - def _prerun_002(self, engine): data = [{'id': 1, 'key': 'fake-key', 'image_path': '/dev/null', 'pxe_config_path': '/dev/null/', 'root_mb': 0, 'swap_mb': 0}]