Remove db locks and use random db names for tests

Removed database locks against the database openstack_citest and
implemented random name generation for new databases created just
for persistence testing.
Once the locks were removed, using 'template1' as the initial db
for postgres connections was problematic, because postgres refuses
to create databases when there are multiple connections to
'template1'.  Switched to using 'postgres' as an initial db to use.
Changed _reset_database to _init_db since we are always working
with a brand new database, and removed the 'drop database' before
a 'create database.
Added a _remove_db method to remove the database once testing was
finished.

Change-Id: Iaf1c101df9c268da48db7432bcbc0467f6486bcd
Closes-Bug: 1327469

1 files changed, 67 insertions, 57 deletions

diff --git a/taskflow/tests/unit/persistence/test_sql_persistence.py b/taskflow/tests/unit/persistence/test_sql_persistence.py
index b48f84a..2baaa37 100644
--- a/taskflow/tests/unit/persistence/test_sql_persistence.py
+++ b/taskflow/tests/unit/persistence/test_sql_persistence.py
@@ -16,8 +16,8 @@
 
 import contextlib
 import os
+import random
 import tempfile
-import threading
 
 import testtools
 
@@ -29,12 +29,14 @@ import testtools
 # There are also "opportunistic" tests for both mysql and postgresql in here,
 # which allows testing against all 3 databases (sqlite, mysql, postgres) in
 # a properly configured unit test environment. For the opportunistic testing
-# you need to set up a db named 'openstack_citest' with user 'openstack_citest'
-# and password 'openstack_citest' on localhost.
+# you need to set up a db user 'openstack_citest' with password
+# 'openstack_citest' that has the permissions to create databases on
+# localhost.
 
 USER = "openstack_citest"
 PASSWD = "openstack_citest"
-DATABASE = "openstack_citest"
+DATABASE = "tftest_" + ''.join(random.choice('0123456789')
+                               for _ in range(12))
 
 try:
     from taskflow.persistence.backends import impl_sqlalchemy
@@ -50,7 +52,6 @@ MYSQL_VARIANTS = ('mysqldb', 'pymysql')
 from taskflow.persistence import backends
 from taskflow import test
 from taskflow.tests.unit.persistence import base
-from taskflow.utils import lock_utils
 
 
 def _get_connect_string(backend, user, passwd, database=None, variant=None):
@@ -97,7 +98,7 @@ def _postgres_exists():
         return False
     engine = None
     try:
-        db_uri = _get_connect_string('postgres', USER, PASSWD, 'template1')
+        db_uri = _get_connect_string('postgres', USER, PASSWD, 'postgres')
         engine = sa.create_engine(db_uri)
         with contextlib.closing(engine.connect()):
             return True
@@ -137,29 +138,31 @@ class SqlitePersistenceTest(test.TestCase, base.PersistenceTestMixin):
 
 class BackendPersistenceTestMixin(base.PersistenceTestMixin):
     """Specifies a backend type and does required setup and teardown."""
-    LOCK_NAME = None
 
     def _get_connection(self):
         return self.backend.get_connection()
 
-    def _reset_database(self):
-        """Resets the database, and returns the uri to that database.
+    def _init_db(self):
+        """Sets up the database, and returns the uri to that database."""
+        raise NotImplementedError()
 
-        Called *only* after locking succeeds.
-        """
+    def _remove_db(self):
+        """Cleans up by removing the database once the tests are done."""
         raise NotImplementedError()
 
     def setUp(self):
         super(BackendPersistenceTestMixin, self).setUp()
         self.backend = None
-        self.big_lock.acquire()
-        self.addCleanup(self.big_lock.release)
         try:
+            self.db_uri = self._init_db()
+            # Since we are using random database names, we need to make sure
+            # and remove our random database when we are done testing.
+            self.addCleanup(self._remove_db)
             conf = {
-                'connection': self._reset_database(),
+                'connection': self.db_uri
             }
         except Exception as e:
-            self.skipTest("Failed to reset your database;"
+            self.skipTest("Failed to create temporary database;"
                           " testing being skipped due to: %s" % (e))
         try:
             self.backend = impl_sqlalchemy.SQLAlchemyBackend(conf)
@@ -174,25 +177,11 @@ class BackendPersistenceTestMixin(base.PersistenceTestMixin):
 @testtools.skipIf(not SQLALCHEMY_AVAILABLE, 'sqlalchemy is not available')
 @testtools.skipIf(not _mysql_exists(), 'mysql is not available')
 class MysqlPersistenceTest(BackendPersistenceTestMixin, test.TestCase):
-    LOCK_NAME = 'mysql_persistence_test'
 
     def __init__(self, *args, **kwargs):
         test.TestCase.__init__(self, *args, **kwargs)
-        # We need to make sure that each test goes through a set of locks
-        # to ensure that multiple tests are not modifying the database,
-        # dropping it, creating it at the same time. To accomplish this we use
-        # a lock that ensures multiple parallel processes can't run at the
-        # same time as well as a in-process lock to ensure that multiple
-        # threads can't run at the same time.
-        lock_path = os.path.join(tempfile.gettempdir(),
-                                 'taskflow-%s.lock' % (self.LOCK_NAME))
-        locks = [
-            lock_utils.InterProcessLock(lock_path),
-            threading.RLock(),
-        ]
-        self.big_lock = lock_utils.MultiLock(locks)
-
-    def _reset_database(self):
+
+    def _init_db(self):
         working_variant = None
         for variant in MYSQL_VARIANTS:
             engine = None
@@ -201,7 +190,6 @@ class MysqlPersistenceTest(BackendPersistenceTestMixin, test.TestCase):
                                              variant=variant)
                 engine = sa.create_engine(db_uri)
                 with contextlib.closing(engine.connect()) as conn:
-                    conn.execute("DROP DATABASE IF EXISTS %s" % DATABASE)
                     conn.execute("CREATE DATABASE %s" % DATABASE)
                     working_variant = variant
             except Exception:
@@ -216,60 +204,82 @@ class MysqlPersistenceTest(BackendPersistenceTestMixin, test.TestCase):
                 break
         if not working_variant:
             variants = ", ".join(MYSQL_VARIANTS)
-            self.skipTest("Failed to find a mysql variant"
-                          " (tried %s) that works; mysql testing"
-                          " being skipped" % (variants))
+            raise Exception("Failed to initialize MySQL db."
+                            "  Tried these variants:  %s; MySQL testing"
+                            " being skipped" % (variants))
         else:
             return _get_connect_string('mysql', USER, PASSWD,
                                        database=DATABASE,
                                        variant=working_variant)
 
+    def _remove_db(self):
+        engine = None
+        try:
+            engine = sa.create_engine(self.db_uri)
+            with contextlib.closing(engine.connect()) as conn:
+                conn.execute("DROP DATABASE IF EXISTS %s" % DATABASE)
+        except Exception as e:
+            raise Exception('Failed to remove temporary database: %s' % (e))
+        finally:
+            if engine is not None:
+                try:
+                    engine.dispose()
+                except Exception:
+                    pass
+
 
 @testtools.skipIf(not SQLALCHEMY_AVAILABLE, 'sqlalchemy is not available')
 @testtools.skipIf(not _postgres_exists(), 'postgres is not available')
 class PostgresPersistenceTest(BackendPersistenceTestMixin, test.TestCase):
-    LOCK_NAME = 'postgres_persistence_test'
 
     def __init__(self, *args, **kwargs):
         test.TestCase.__init__(self, *args, **kwargs)
-        # We need to make sure that each test goes through a set of locks
-        # to ensure that multiple tests are not modifying the database,
-        # dropping it, creating it at the same time. To accomplish this we use
-        # a lock that ensures multiple parallel processes can't run at the
-        # same time as well as a in-process lock to ensure that multiple
-        # threads can't run at the same time.
-        lock_path = os.path.join(tempfile.gettempdir(),
-                                 'taskflow-%s.lock' % (self.LOCK_NAME))
-        locks = [
-            lock_utils.InterProcessLock(lock_path),
-            threading.RLock(),
-        ]
-        self.big_lock = lock_utils.MultiLock(locks)
-
-    def _reset_database(self):
+
+    def _init_db(self):
         engine = None
         try:
             # Postgres can't operate on the database it's connected to, that's
-            # why we connect to the default template database 'template1' and
-            # then drop and create the desired database.
+            # why we connect to the database 'postgres' and then create the
+            # desired database.
             db_uri = _get_connect_string('postgres', USER, PASSWD,
-                                         database='template1')
+                                         database='postgres')
             engine = sa.create_engine(db_uri)
             with contextlib.closing(engine.connect()) as conn:
                 conn.connection.set_isolation_level(0)
-                conn.execute("DROP DATABASE IF EXISTS %s" % DATABASE)
+                conn.execute("CREATE DATABASE %s" % DATABASE)
                 conn.connection.set_isolation_level(1)
+        except Exception as e:
+            raise Exception('Failed to initialize PostgreSQL db: %s' % (e))
+        finally:
+            if engine is not None:
+                try:
+                    engine.dispose()
+                except Exception:
+                    pass
+        return _get_connect_string('postgres', USER, PASSWD,
+                                   database=DATABASE)
+
+    def _remove_db(self):
+        engine = None
+        try:
+            # Postgres can't operate on the database it's connected to, that's
+            # why we connect to the 'postgres' database and then drop the
+            # database.
+            db_uri = _get_connect_string('postgres', USER, PASSWD,
+                                         database='postgres')
+            engine = sa.create_engine(db_uri)
             with contextlib.closing(engine.connect()) as conn:
                 conn.connection.set_isolation_level(0)
-                conn.execute("CREATE DATABASE %s" % DATABASE)
+                conn.execute("DROP DATABASE IF EXISTS %s" % DATABASE)
                 conn.connection.set_isolation_level(1)
+        except Exception as e:
+            raise Exception('Failed to remove temporary database: %s' % (e))
         finally:
             if engine is not None:
                 try:
                     engine.dispose()
                 except Exception:
                     pass
-        return _get_connect_string('postgres', USER, PASSWD, database=DATABASE)
 
 
 @testtools.skipIf(not SQLALCHEMY_AVAILABLE, 'sqlalchemy is not available')