diff --git a/taskflow/tests/unit/persistence/test_dir_persistence.py b/taskflow/tests/unit/persistence/test_dir_persistence.py
index 076ab2d4..ee40635a 100644
--- a/taskflow/tests/unit/persistence/test_dir_persistence.py
+++ b/taskflow/tests/unit/persistence/test_dir_persistence.py
@@ -16,10 +16,12 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import contextlib
 import os
 import shutil
 import tempfile
 
+from taskflow.persistence import backends
 from taskflow.persistence.backends import impl_dir
 from taskflow import test
 from taskflow.tests.unit.persistence import base
@@ -45,3 +47,20 @@ class DirPersistenceTest(test.TestCase, base.PersistenceTestMixin):
         if self.path and os.path.isdir(self.path):
             shutil.rmtree(self.path)
         self.path = None
+
+    def test_dir_persistence_entry_point(self):
+        conf = {
+            'connection': 'dir:',
+            'path': self.path
+        }
+        backend = backends.fetch(conf)
+        self.assertIsInstance(backend, impl_dir.DirBackend)
+        backend.close()
+
+    def test_file_persistence_entry_point(self):
+        conf = {
+            'connection': 'file:',
+            'path': self.path
+        }
+        with contextlib.closing(backends.fetch(conf)) as be:
+            self.assertIsInstance(be, impl_dir.DirBackend)
diff --git a/taskflow/tests/unit/persistence/test_memory_persistence.py b/taskflow/tests/unit/persistence/test_memory_persistence.py
index cccf9fbc..111c8c33 100644
--- a/taskflow/tests/unit/persistence/test_memory_persistence.py
+++ b/taskflow/tests/unit/persistence/test_memory_persistence.py
@@ -16,6 +16,9 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import contextlib
+
+from taskflow.persistence import backends
 from taskflow.persistence.backends import impl_memory
 from taskflow import test
 from taskflow.tests.unit.persistence import base
@@ -34,3 +37,8 @@ class MemoryPersistenceTest(test.TestCase, base.PersistenceTestMixin):
         conn.clear_all()
         self._backend = None
         super(MemoryPersistenceTest, self).tearDown()
+
+    def test_memory_persistence_entry_point(self):
+        conf = {'connection': 'memory:'}
+        with contextlib.closing(backends.fetch(conf)) as be:
+            self.assertIsInstance(be, impl_memory.MemoryBackend)
diff --git a/taskflow/tests/unit/persistence/test_sql_persistence.py b/taskflow/tests/unit/persistence/test_sql_persistence.py
index 2306d0d9..d3147457 100644
--- a/taskflow/tests/unit/persistence/test_sql_persistence.py
+++ b/taskflow/tests/unit/persistence/test_sql_persistence.py
@@ -50,6 +50,7 @@ except ImportError:
 # Testing will try to run against these two mysql library variants.
 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
@@ -261,3 +262,26 @@ class PostgresPersistenceTest(BackendPersistenceTestMixin, test.TestCase):
             if engine is not None:
                 engine.dispose()
         return _get_connect_string('postgres', USER, PASSWD, database=DATABASE)
+
+
+@testtools.skipIf(not SQLALCHEMY_AVAILABLE, 'sqlalchemy is not available')
+class SQLBackendFetchingTest(test.TestCase):
+
+    def test_sqlite_persistence_entry_point(self):
+        conf = {'connection': 'sqlite:///'}
+        with contextlib.closing(backends.fetch(conf)) as be:
+            self.assertIsInstance(be, impl_sqlalchemy.SQLAlchemyBackend)
+
+    @testtools.skipIf(not _postgres_exists(), 'postgres is not available')
+    def test_mysql_persistence_entry_point(self):
+        uri = "mysql://%s:%s@localhost/%s" % (USER, PASSWD, DATABASE)
+        conf = {'connection': uri}
+        with contextlib.closing(backends.fetch(conf)) as be:
+            self.assertIsInstance(be, impl_sqlalchemy.SQLAlchemyBackend)
+
+    @testtools.skipIf(not _mysql_exists(), 'mysql is not available')
+    def test_postgres_persistence_entry_point(self):
+        uri = "postgresql://%s:%s@localhost/%s" % (USER, PASSWD, DATABASE)
+        conf = {'connection': uri}
+        with contextlib.closing(backends.fetch(conf)) as be:
+            self.assertIsInstance(be, impl_sqlalchemy.SQLAlchemyBackend)
diff --git a/taskflow/tests/unit/persistence/test_zake_persistence.py b/taskflow/tests/unit/persistence/test_zake_persistence.py
index 1de5267b..22e8ab5e 100644
--- a/taskflow/tests/unit/persistence/test_zake_persistence.py
+++ b/taskflow/tests/unit/persistence/test_zake_persistence.py
@@ -16,6 +16,9 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import contextlib
+
+from taskflow.persistence import backends
 from taskflow.persistence.backends import impl_zookeeper
 from taskflow import test
 from taskflow.tests.unit.persistence import base
@@ -38,3 +41,8 @@ class ZkPersistenceTest(test.TestCase, base.PersistenceTestMixin):
         super(ZkPersistenceTest, self).tearDown()
         conn = self._get_connection()
         conn.clear_all()
+
+    def test_zk_persistence_entry_point(self):
+        conf = {'connection': 'zookeeper:'}
+        with contextlib.closing(backends.fetch(conf)) as be:
+            self.assertIsInstance(be, impl_zookeeper.ZkBackend)