diff --git a/manila_tempest_tests/tests/api/admin/test_migration.py b/manila_tempest_tests/tests/api/admin/test_migration.py
index 1bc67fb..e842ead 100644
--- a/manila_tempest_tests/tests/api/admin/test_migration.py
+++ b/manila_tempest_tests/tests/api/admin/test_migration.py
@@ -450,7 +450,6 @@
             new_share_network_id = self.provide_share_network(
                 self.shares_v2_client,
                 self.networks_client,
-                isolated_creds_client=None,
                 ignore_multitenancy_config=True,
             )
         else:
diff --git a/manila_tempest_tests/tests/api/admin/test_quotas.py b/manila_tempest_tests/tests/api/admin/test_quotas.py
index 215b25d..116180e 100644
--- a/manila_tempest_tests/tests/api/admin/test_quotas.py
+++ b/manila_tempest_tests/tests/api/admin/test_quotas.py
@@ -39,15 +39,16 @@
             msg = "Quota tests are disabled."
             raise cls.skipException(msg)
         super(SharesAdminQuotasTest, cls).resource_setup()
-        cls.user_id = cls.shares_v2_client.user_id
-        cls.tenant_id = cls.shares_v2_client.tenant_id
+        cls.client = cls.shares_v2_client
+        cls.user_id = cls.client.user_id
+        cls.tenant_id = cls.client.tenant_id
         # create share type
         cls.share_type = cls._create_share_type()
         cls.share_type_id = cls.share_type['id']
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_default_quotas(self):
-        quotas = self.shares_v2_client.default_quotas(self.tenant_id)
+        quotas = self.client.default_quotas(self.tenant_id)
         self.assertGreater(int(quotas["gigabytes"]), -2)
         self.assertGreater(int(quotas["snapshot_gigabytes"]), -2)
         self.assertGreater(int(quotas["shares"]), -2)
@@ -62,7 +63,7 @@
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_show_quotas(self):
-        quotas = self.shares_v2_client.show_quotas(self.tenant_id)
+        quotas = self.client.show_quotas(self.tenant_id)
         self.assertGreater(int(quotas["gigabytes"]), -2)
         self.assertGreater(int(quotas["snapshot_gigabytes"]), -2)
         self.assertGreater(int(quotas["shares"]), -2)
@@ -77,7 +78,7 @@
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_show_quotas_for_user(self):
-        quotas = self.shares_v2_client.show_quotas(
+        quotas = self.client.show_quotas(
             self.tenant_id, self.user_id)
         self.assertGreater(int(quotas["gigabytes"]), -2)
         self.assertGreater(int(quotas["snapshot_gigabytes"]), -2)
@@ -94,7 +95,7 @@
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     @base.skip_if_microversion_not_supported(PRE_SHARE_GROUPS_MICROVERSION)
     def test_show_sg_quotas_using_too_old_microversion(self):
-        quotas = self.shares_v2_client.show_quotas(
+        quotas = self.client.show_quotas(
             self.tenant_id, version=PRE_SHARE_GROUPS_MICROVERSION)
 
         for key in ('share_groups', 'share_group_snapshots'):
@@ -103,7 +104,7 @@
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     @base.skip_if_microversion_not_supported(PRE_SHARE_GROUPS_MICROVERSION)
     def test_show_sg_quotas_for_user_using_too_old_microversion(self):
-        quotas = self.shares_v2_client.show_quotas(
+        quotas = self.client.show_quotas(
             self.tenant_id, self.user_id,
             version=PRE_SHARE_GROUPS_MICROVERSION)
 
@@ -114,11 +115,11 @@
     @base.skip_if_microversion_not_supported(
         PRE_SHARE_REPLICA_QUOTAS_MICROVERSION)
     def test_show_replica_quotas_for_user_using_too_old_microversion(self):
-        quotas = self.shares_v2_client.show_quotas(
+        quotas = self.client.show_quotas(
             self.tenant_id, self.user_id,
             version=PRE_SHARE_REPLICA_QUOTAS_MICROVERSION)
 
-        for key in('share_replicas', 'replica_gigabytes'):
+        for key in ('share_replicas', 'replica_gigabytes'):
             self.assertNotIn(key, quotas)
 
     @ddt.data(
@@ -150,10 +151,10 @@
             keys.append('replica_gigabytes')
 
         # Get current project quotas
-        p_quotas = self.shares_v2_client.show_quotas(self.tenant_id)
+        p_quotas = self.client.show_quotas(self.tenant_id)
 
-        # Get current quotas
-        st_quotas = self.shares_v2_client.show_quotas(
+        # Get current share type quotas
+        st_quotas = self.client.show_quotas(
             self.tenant_id, share_type=share_type[share_type_key])
 
         # Share type quotas have values equal to project's
@@ -169,13 +170,22 @@
 @ddt.ddt
 class SharesAdminQuotasUpdateTest(base.BaseSharesAdminTest):
 
+    # We want to force a fresh project for this test class, since we'll be
+    # manipulating project quotas - and any pre-existing projects may have
+    # resources, quotas and the like that might interfere with our test cases.
     force_tenant_isolation = True
 
     @classmethod
-    def resource_setup(cls):
+    def skip_checks(cls):
+        super(SharesAdminQuotasUpdateTest, cls).skip_checks()
+        if not CONF.auth.use_dynamic_credentials:
+            raise cls.skipException('Dynamic credentials are required')
         if not CONF.share.run_quota_tests:
             msg = "Quota tests are disabled."
             raise cls.skipException(msg)
+
+    @classmethod
+    def resource_setup(cls):
         super(SharesAdminQuotasUpdateTest, cls).resource_setup()
         # create share type
         cls.share_type = cls._create_share_type()
@@ -183,12 +193,9 @@
         # create share group type
         cls.share_group_type = cls._create_share_group_type()
         cls.share_group_type_id = cls.share_group_type['id']
-
-    def setUp(self):
-        super(self.__class__, self).setUp()
-        self.client = self.get_client_with_isolated_creds(client_version='2')
-        self.tenant_id = self.client.tenant_id
-        self.user_id = self.client.user_id
+        cls.client = cls.shares_v2_client
+        cls.tenant_id = cls.client.tenant_id
+        cls.user_id = cls.client.user_id
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_update_tenant_quota_shares(self):
@@ -197,7 +204,8 @@
         new_quota = int(quotas["shares"]) + 2
 
         # set new quota for shares
-        updated = self.client.update_quotas(self.tenant_id, shares=new_quota)
+        updated = self.update_quotas(self.tenant_id, shares=new_quota)
+
         self.assertEqual(new_quota, int(updated["shares"]))
 
     @ddt.data(
@@ -214,8 +222,8 @@
         new_quota = int(quotas[quota_key]) + 2
 
         # Set new quota
-        updated = self.client.update_quotas(
-            self.tenant_id, **{quota_key: new_quota})
+        updated = self.update_quotas(self.tenant_id, **{quota_key: new_quota})
+
         self.assertEqual(new_quota, int(updated[quota_key]))
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
@@ -225,8 +233,10 @@
         new_quota = int(quotas["shares"]) - 1
 
         # set new quota for shares
-        updated = self.client.update_quotas(
-            self.tenant_id, self.user_id, shares=new_quota)
+        updated = self.update_quotas(self.tenant_id,
+                                     user_id=self.user_id,
+                                     shares=new_quota)
+
         self.assertEqual(new_quota, int(updated["shares"]))
 
     @ddt.data(
@@ -243,8 +253,10 @@
         new_quota = int(quotas[quota_key]) - 1
 
         # Set new quota
-        updated = self.client.update_quotas(
-            self.tenant_id, self.user_id, **{quota_key: new_quota})
+        updated = self.update_quotas(self.tenant_id,
+                                     user_id=self.user_id,
+                                     **{quota_key: new_quota})
+
         self.assertEqual(new_quota, int(updated[quota_key]))
 
     @ddt.data(("share_replicas", False),
@@ -268,8 +280,8 @@
         kwargs.update({quota_key: new_quota})
 
         # Set the new quota based on tenant or tenant and user_id
-        updated = self.client.update_quotas(
-            self.tenant_id, **kwargs)
+        updated = self.update_quotas(self.tenant_id, **kwargs)
+
         self.assertEqual(new_quota, int(updated[quota_key]))
 
     @ddt.data(
@@ -283,7 +295,7 @@
         # Check if the used microversion supports 'share_replica' and
         # 'replica_gigabytes' quotas
         replica_quotas_supported = utils.share_replica_quotas_are_supported()
-        share_type = self._create_share_type()
+        share_type = self._create_share_type(is_public=is_st_public)
 
         # Get current quotas
         quotas = self.client.show_quotas(
@@ -298,10 +310,14 @@
         for q in quota_keys:
             new_quota = int(quotas[q]) - 1
 
-            # Set new quota
-            updated = self.client.update_quotas(
-                self.tenant_id, share_type=share_type[share_type_key],
-                **{q: new_quota})
+            # Set new quota, cleanup isn't necessary, share type will be
+            # deleted when the test concludes, and that'll take care of
+            # share type quotas
+            updated = self.update_quotas(self.tenant_id,
+                                         share_type=share_type[share_type_key],
+                                         cleanup=False,
+                                         **{q: new_quota})
+
             self.assertEqual(new_quota, int(updated[q]))
 
         current_quotas = self.client.show_quotas(
@@ -317,8 +333,8 @@
         new_quota = int(quotas["snapshots"]) + 2
 
         # set new quota for snapshots
-        updated = self.client.update_quotas(
-            self.tenant_id, snapshots=new_quota)
+        updated = self.update_quotas(self.tenant_id, snapshots=new_quota)
+
         self.assertEqual(new_quota, int(updated["snapshots"]))
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
@@ -328,8 +344,10 @@
         new_quota = int(quotas["snapshots"]) - 1
 
         # set new quota for snapshots
-        updated = self.client.update_quotas(
-            self.tenant_id, self.user_id, snapshots=new_quota)
+        updated = self.update_quotas(self.tenant_id,
+                                     user_id=self.user_id,
+                                     snapshots=new_quota)
+
         self.assertEqual(new_quota, int(updated["snapshots"]))
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
@@ -341,8 +359,8 @@
         gigabytes = int(custom["gigabytes"]) + 2
 
         # set new quota for shares
-        updated = self.client.update_quotas(
-            self.tenant_id, gigabytes=gigabytes)
+        updated = self.update_quotas(self.tenant_id, gigabytes=gigabytes)
+
         self.assertEqual(gigabytes, int(updated["gigabytes"]))
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
@@ -354,9 +372,9 @@
         snapshot_gigabytes = int(custom["snapshot_gigabytes"]) + 2
 
         # set new quota for shares
-        updated = self.client.update_quotas(
-            self.tenant_id,
-            snapshot_gigabytes=snapshot_gigabytes)
+        updated = self.update_quotas(self.tenant_id,
+                                     snapshot_gigabytes=snapshot_gigabytes)
+
         self.assertEqual(snapshot_gigabytes,
                          int(updated["snapshot_gigabytes"]))
 
@@ -369,8 +387,10 @@
         gigabytes = int(custom["gigabytes"]) - 1
 
         # set new quota for shares
-        updated = self.client.update_quotas(
-            self.tenant_id, self.user_id, gigabytes=gigabytes)
+        updated = self.update_quotas(self.tenant_id,
+                                     user_id=self.user_id,
+                                     gigabytes=gigabytes)
+
         self.assertEqual(gigabytes, int(updated["gigabytes"]))
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
@@ -382,9 +402,10 @@
         snapshot_gigabytes = int(custom["snapshot_gigabytes"]) - 1
 
         # set new quota for shares
-        updated = self.client.update_quotas(
-            self.tenant_id, self.user_id,
-            snapshot_gigabytes=snapshot_gigabytes)
+        updated = self.update_quotas(self.tenant_id,
+                                     user_id=self.user_id,
+                                     snapshot_gigabytes=snapshot_gigabytes)
+
         self.assertEqual(snapshot_gigabytes,
                          int(updated["snapshot_gigabytes"]))
 
@@ -395,8 +416,8 @@
         new_quota = int(quotas["share_networks"]) + 2
 
         # set new quota for share-networks
-        updated = self.client.update_quotas(
-            self.tenant_id, share_networks=new_quota)
+        updated = self.update_quotas(self.tenant_id, share_networks=new_quota)
+
         self.assertEqual(new_quota, int(updated["share_networks"]))
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
@@ -407,9 +428,10 @@
         new_quota = int(quotas["share_networks"]) - 1
 
         # set new quota for share-networks
-        updated = self.client.update_quotas(
-            self.tenant_id, self.user_id,
-            share_networks=new_quota)
+        updated = self.update_quotas(self.tenant_id,
+                                     user_id=self.user_id,
+                                     share_networks=new_quota)
+
         self.assertEqual(new_quota, int(updated["share_networks"]))
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
@@ -437,8 +459,8 @@
             data["share_replicas"] = int(custom["share_replicas"]) + 2
             data["replica_gigabytes"] = int(custom["replica_gigabytes"]) + 2
 
-        # set new quota
-        updated = self.client.update_quotas(self.tenant_id, **data)
+        # set new quota, turn off cleanup - we'll do it right below
+        updated = self.update_quotas(self.tenant_id, cleanup=False, **data)
         self.assertEqual(data["shares"], int(updated["shares"]))
         self.assertEqual(data["snapshots"], int(updated["snapshots"]))
         self.assertEqual(data["gigabytes"], int(updated["gigabytes"]))
@@ -503,7 +525,7 @@
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     @base.skip_if_microversion_lt("2.39")
     def test_reset_share_type_quotas(self, share_type_key, is_st_public):
-        share_type = self._create_share_type()
+        share_type = self._create_share_type(is_public=is_st_public)
         quota_keys = ['shares', 'snapshots', 'gigabytes', 'snapshot_gigabytes']
 
         # get default_quotas
@@ -519,8 +541,8 @@
             quota_keys.append('share_replicas')
             quota_keys.append('replica_gigabytes')
 
-        # set new quota for project
-        updated_p_quota = self.client.update_quotas(
+        # set new project quota
+        updated_p_quota = self.update_quotas(
             self.tenant_id,
             shares=int(default_quotas['shares']) + 5,
             snapshots=int(default_quotas['snapshots']) + 5,
@@ -533,14 +555,15 @@
             kwargs.update(self._get_new_replica_quota_values(
                 default_quotas, 3))
 
-        # set new quota for project
-        self.client.update_quotas(
+        # set share type quota for project, don't cleanup, we'll do that below
+        self.update_quotas(
             self.tenant_id,
             share_type=share_type[share_type_key],
             shares=int(default_quotas['shares']) + 3,
             snapshots=int(default_quotas['snapshots']) + 3,
             gigabytes=int(default_quotas['gigabytes']) + 3,
             snapshot_gigabytes=int(default_quotas['snapshot_gigabytes']) + 3,
+            cleanup=False,
             **kwargs
         )
 
@@ -561,7 +584,7 @@
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_unlimited_quota_for_shares(self):
-        self.client.update_quotas(self.tenant_id, shares=-1)
+        self.update_quotas(self.tenant_id, shares=-1)
 
         quotas = self.client.show_quotas(self.tenant_id)
 
@@ -569,8 +592,7 @@
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_unlimited_user_quota_for_shares(self):
-        self.client.update_quotas(
-            self.tenant_id, self.user_id, shares=-1)
+        self.update_quotas(self.tenant_id, user_id=self.user_id, shares=-1)
 
         quotas = self.client.show_quotas(self.tenant_id, self.user_id)
 
@@ -578,7 +600,7 @@
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_unlimited_quota_for_snapshots(self):
-        self.client.update_quotas(self.tenant_id, snapshots=-1)
+        self.update_quotas(self.tenant_id, snapshots=-1)
 
         quotas = self.client.show_quotas(self.tenant_id)
 
@@ -586,8 +608,7 @@
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_unlimited_user_quota_for_snapshots(self):
-        self.client.update_quotas(
-            self.tenant_id, self.user_id, snapshots=-1)
+        self.update_quotas(self.tenant_id, user_id=self.user_id, snapshots=-1)
 
         quotas = self.client.show_quotas(self.tenant_id, self.user_id)
 
@@ -595,7 +616,7 @@
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_unlimited_quota_for_gigabytes(self):
-        self.client.update_quotas(self.tenant_id, gigabytes=-1)
+        self.update_quotas(self.tenant_id, gigabytes=-1)
 
         quotas = self.client.show_quotas(self.tenant_id)
 
@@ -603,8 +624,7 @@
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_unlimited_quota_for_snapshot_gigabytes(self):
-        self.client.update_quotas(
-            self.tenant_id, snapshot_gigabytes=-1)
+        self.update_quotas(self.tenant_id, snapshot_gigabytes=-1)
 
         quotas = self.client.show_quotas(self.tenant_id)
 
@@ -612,8 +632,7 @@
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_unlimited_user_quota_for_gigabytes(self):
-        self.client.update_quotas(
-            self.tenant_id, self.user_id, gigabytes=-1)
+        self.update_quotas(self.tenant_id, user_id=self.user_id, gigabytes=-1)
 
         quotas = self.client.show_quotas(self.tenant_id, self.user_id)
 
@@ -621,8 +640,9 @@
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_unlimited_user_quota_for_snapshot_gigabytes(self):
-        self.client.update_quotas(
-            self.tenant_id, self.user_id, snapshot_gigabytes=-1)
+        self.update_quotas(self.tenant_id,
+                           user_id=self.user_id,
+                           snapshot_gigabytes=-1)
 
         quotas = self.client.show_quotas(self.tenant_id, self.user_id)
 
@@ -630,7 +650,7 @@
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_unlimited_quota_for_share_networks(self):
-        self.client.update_quotas(self.tenant_id, share_networks=-1)
+        self.update_quotas(self.tenant_id, share_networks=-1)
 
         quotas = self.client.show_quotas(self.tenant_id)
 
@@ -638,8 +658,9 @@
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_unlimited_user_quota_for_share_networks(self):
-        self.client.update_quotas(
-            self.tenant_id, self.user_id, share_networks=-1)
+        self.update_quotas(self.tenant_id,
+                           user_id=self.user_id,
+                           share_networks=-1)
 
         quotas = self.client.show_quotas(self.tenant_id, self.user_id)
 
@@ -650,7 +671,7 @@
         CONF.share.run_share_group_tests, 'Share Group tests disabled.')
     @utils.skip_if_microversion_not_supported(SHARE_GROUPS_MICROVERSION)
     def test_unlimited_quota_for_share_groups(self):
-        self.client.update_quotas(self.tenant_id, share_groups=-1)
+        self.update_quotas(self.tenant_id, share_groups=-1)
 
         quotas = self.client.show_quotas(self.tenant_id)
 
@@ -661,8 +682,9 @@
         CONF.share.run_share_group_tests, 'Share Group tests disabled.')
     @utils.skip_if_microversion_not_supported(SHARE_GROUPS_MICROVERSION)
     def test_unlimited_user_quota_for_share_group_snapshots(self):
-        self.client.update_quotas(
-            self.tenant_id, self.user_id, share_group_snapshots=-1)
+        self.update_quotas(self.tenant_id,
+                           user_id=self.user_id,
+                           share_group_snapshots=-1)
 
         quotas = self.client.show_quotas(self.tenant_id, self.user_id)
 
@@ -674,7 +696,7 @@
         SHARE_REPLICA_QUOTAS_MICROVERSION)
     def test_unlimited_quota_for_replica_quotas(self, quota_key):
         kwargs = {quota_key: -1}
-        self.client.update_quotas(self.tenant_id, **kwargs)
+        self.update_quotas(self.tenant_id, **kwargs)
 
         quotas = self.client.show_quotas(self.tenant_id)
 
@@ -686,7 +708,7 @@
         SHARE_REPLICA_QUOTAS_MICROVERSION)
     def test_unlimited_user_quota_for_replica_quotas(self, quota_key):
         kwargs = {quota_key: -1}
-        self.client.update_quotas(self.tenant_id, self.user_id, **kwargs)
+        self.update_quotas(self.tenant_id, user_id=self.user_id, **kwargs)
 
         quotas = self.client.show_quotas(self.tenant_id, self.user_id)
 
@@ -695,35 +717,49 @@
     @ddt.data(11, -1)
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_update_user_quotas_bigger_than_project_quota(self, user_quota):
-        self.client.update_quotas(self.tenant_id, shares=10)
-        self.client.update_quotas(
-            self.tenant_id, user_id=self.user_id, force=True,
-            shares=user_quota)
+        self.update_quotas(self.tenant_id, shares=10)
+
+        self.update_quotas(self.tenant_id,
+                           user_id=self.user_id,
+                           force=True,
+                           shares=user_quota)
 
     @ddt.data(11, -1)
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     @base.skip_if_microversion_lt("2.39")
     def test_update_share_type_quotas_bigger_than_project_quota(self, st_q):
         share_type = self._create_share_type()
-        self.client.update_quotas(self.tenant_id, shares=10)
 
-        self.client.update_quotas(
-            self.tenant_id, share_type=share_type['name'], force=True,
-            shares=st_q)
+        self.update_quotas(self.tenant_id, shares=10)
+
+        # no need to cleanup share type quota, share type will be deleted at
+        # the end of the test
+        self.update_quotas(self.tenant_id,
+                           share_type=share_type['name'],
+                           force=True,
+                           cleanup=False,
+                           shares=st_q)
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     @base.skip_if_microversion_lt("2.39")
     def test_set_share_type_quota_bigger_than_users_quota(self):
         share_type = self._create_share_type()
-        self.client.update_quotas(self.tenant_id, force=False, shares=13)
-        self.client.update_quotas(
-            self.tenant_id, user_id=self.user_id, force=False, shares=11)
+
+        self.update_quotas(self.tenant_id, force=False, shares=13)
+
+        self.update_quotas(self.tenant_id,
+                           user_id=self.user_id,
+                           force=True,
+                           shares=11)
 
         # Share type quota does not depend on user's quota, so we should be
-        # able to update it.
-        self.client.update_quotas(
-            self.tenant_id, share_type=share_type['name'], force=False,
-            shares=12)
+        # able to update it. No need for cleanup, since the share type will
+        # be deleted when the test completes, cleaning up quotas and usages
+        self.update_quotas(self.tenant_id,
+                           share_type=share_type['name'],
+                           force=False,
+                           cleanup=False,
+                           shares=12)
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
     @base.skip_if_microversion_lt("2.39")
@@ -733,12 +769,21 @@
                       for i in (1, 2))
 
         # Set quotas for project, user and both share types
-        self.client.update_quotas(self.tenant_id, shares=3, gigabytes=10)
-        self.client.update_quotas(
-            self.tenant_id, user_id=self.user_id, shares=2, gigabytes=7)
+        self.update_quotas(self.tenant_id, shares=3, gigabytes=10)
+
+        self.update_quotas(self.tenant_id,
+                           user_id=self.user_id,
+                           shares=2,
+                           gigabytes=7)
+
         for st in (st_1['id'], st_2['name']):
-            self.client.update_quotas(
-                self.tenant_id, share_type=st, shares=2, gigabytes=4)
+            # no need for cleanup, since share types will be deleted at the
+            # end of the test
+            self.update_quotas(self.tenant_id,
+                               share_type=st,
+                               shares=2,
+                               gigabytes=4,
+                               cleanup=False)
 
         # Create share, 4Gb, st1 - ok
         share_1 = self.create_share(
@@ -814,11 +859,14 @@
     @base.skip_if_microversion_lt(SHARE_GROUPS_MICROVERSION)
     def test_share_group_quotas_usages(self):
         # Set quotas for project (3 SG, 1 SGS) and user (2 SG, 1 SGS)
-        self.client.update_quotas(
-            self.tenant_id, share_groups=3, share_group_snapshots=1)
-        self.client.update_quotas(
-            self.tenant_id, user_id=self.user_id,
-            share_groups=2, share_group_snapshots=1)
+        self.update_quotas(self.tenant_id,
+                           share_groups=3,
+                           share_group_snapshots=1)
+
+        self.update_quotas(self.tenant_id,
+                           user_id=self.user_id,
+                           share_groups=2,
+                           share_group_snapshots=1)
 
         # Check usages, they should be 0s
         self._check_usages(0, 0)
diff --git a/manila_tempest_tests/tests/api/admin/test_quotas_negative.py b/manila_tempest_tests/tests/api/admin/test_quotas_negative.py
index ae5e227..7d81615 100644
--- a/manila_tempest_tests/tests/api/admin/test_quotas_negative.py
+++ b/manila_tempest_tests/tests/api/admin/test_quotas_negative.py
@@ -33,16 +33,26 @@
 @ddt.ddt
 class SharesAdminQuotasNegativeTest(base.BaseSharesAdminTest):
 
+    # We want to force some fresh projects for this test class, since we'll be
+    # manipulating project quotas - and any pre-existing projects may have
+    # resources, quotas and the like that might interfere with our test cases.
     force_tenant_isolation = True
 
     @classmethod
-    def resource_setup(cls):
+    def skip_checks(cls):
+        super(SharesAdminQuotasNegativeTest, cls).skip_checks()
+        if not CONF.auth.use_dynamic_credentials:
+            raise cls.skipException('Dynamic credentials are required')
         if not CONF.share.run_quota_tests:
             msg = "Quota tests are disabled."
             raise cls.skipException(msg)
+
+    @classmethod
+    def resource_setup(cls):
         super(SharesAdminQuotasNegativeTest, cls).resource_setup()
-        cls.user_id = cls.shares_client.user_id
-        cls.tenant_id = cls.shares_client.tenant_id
+        cls.client = cls.shares_v2_client
+        cls.user_id = cls.client.user_id
+        cls.tenant_id = cls.client.tenant_id
         # create share type
         cls.share_type = cls._create_share_type()
         cls.share_type_id = cls.share_type['id']
@@ -53,13 +63,12 @@
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     def test_get_quotas_with_empty_tenant_id(self):
         self.assertRaises(lib_exc.NotFound,
-                          self.shares_client.show_quotas, "")
+                          self.client.show_quotas, "")
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     def test_reset_quotas_with_empty_tenant_id(self):
-        client = self.get_client_with_isolated_creds()
         self.assertRaises(lib_exc.NotFound,
-                          client.reset_quotas, "")
+                          self.client.reset_quotas, "")
 
     @ddt.data(
         {"shares": -2},
@@ -71,10 +80,10 @@
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     def test_update_quota_with_wrong_data(self, kwargs):
         # -1 is acceptable value as unlimited
-        client = self.get_client_with_isolated_creds()
-        self.assertRaises(
-            lib_exc.BadRequest,
-            client.update_quotas, client.tenant_id, **kwargs)
+        self.assertRaises(lib_exc.BadRequest,
+                          self.update_quotas,
+                          self.tenant_id,
+                          **kwargs)
 
     @ddt.data(
         {"share_groups": -2},
@@ -86,10 +95,10 @@
     @utils.skip_if_microversion_not_supported(SHARE_GROUPS_MICROVERSION)
     def test_update_sg_quota_with_wrong_data(self, kwargs):
         # -1 is acceptable value as unlimited
-        client = self.get_client_with_isolated_creds(client_version='2')
-        self.assertRaises(
-            lib_exc.BadRequest,
-            client.update_quotas, client.tenant_id, **kwargs)
+        self.assertRaises(lib_exc.BadRequest,
+                          self.update_quotas,
+                          self.tenant_id,
+                          **kwargs)
 
     @ddt.data(
         {"share_replicas": -2},
@@ -100,16 +109,14 @@
         SHARE_REPLICA_QUOTAS_MICROVERSION)
     def test_update_replica_quotas_wrong_data(self, kwargs):
         # -1 is acceptable value as unlimited
-
-        client = self.get_client_with_isolated_creds(client_version='2')
-        self.assertRaises(
-            lib_exc.BadRequest,
-            client.update_quotas, client.tenant_id, **kwargs)
+        self.assertRaises(lib_exc.BadRequest,
+                          self.update_quotas,
+                          self.tenant_id,
+                          **kwargs)
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     def test_create_share_with_size_bigger_than_quota(self):
-        quotas = self.shares_client.show_quotas(
-            self.shares_client.tenant_id)
+        quotas = self.client.show_quotas(self.tenant_id)
         overquota = int(quotas['gigabytes']) + 2
 
         # try schedule share with size, bigger than gigabytes quota
@@ -122,95 +129,86 @@
         CONF.share.run_share_group_tests, 'Share Group tests disabled.')
     @utils.skip_if_microversion_not_supported(SHARE_GROUPS_MICROVERSION)
     def test_create_share_group_with_exceeding_quota_limit(self):
-        client = self.get_client_with_isolated_creds(client_version='2')
-        client.update_quotas(client.tenant_id, share_groups=0)
+        self.update_quotas(self.tenant_id, share_groups=0)
 
         # Try schedule share group creation
-        self.assertRaises(
-            lib_exc.OverLimit,
-            self.create_share_group,
-            share_group_type_id=self.share_group_type_id,
-            share_type_ids=[self.share_type_id],
-            client=client,
-            cleanup_in_class=False)
+        self.assertRaises(lib_exc.OverLimit,
+                          self.create_share_group,
+                          share_group_type_id=self.share_group_type_id,
+                          share_type_ids=[self.share_type_id],
+                          cleanup_in_class=False)
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     def test_try_set_user_quota_shares_bigger_than_tenant_quota(self):
-        client = self.get_client_with_isolated_creds()
 
         # get current quotas for tenant
-        tenant_quotas = client.show_quotas(client.tenant_id)
+        tenant_quotas = self.client.show_quotas(self.tenant_id)
 
         # try set user quota for shares bigger than tenant quota
         bigger_value = int(tenant_quotas["shares"]) + 2
         self.assertRaises(lib_exc.BadRequest,
-                          client.update_quotas,
-                          client.tenant_id,
-                          client.user_id,
+                          self.update_quotas,
+                          self.tenant_id,
+                          user_id=self.user_id,
                           force=False,
                           shares=bigger_value)
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     def test_try_set_user_quota_snaps_bigger_than_tenant_quota(self):
-        client = self.get_client_with_isolated_creds()
 
         # get current quotas for tenant
-        tenant_quotas = client.show_quotas(client.tenant_id)
+        tenant_quotas = self.client.show_quotas(self.tenant_id)
 
         # try set user quota for snapshots bigger than tenant quota
         bigger_value = int(tenant_quotas["snapshots"]) + 2
         self.assertRaises(lib_exc.BadRequest,
-                          client.update_quotas,
-                          client.tenant_id,
-                          client.user_id,
+                          self.update_quotas,
+                          self.tenant_id,
+                          user_id=self.user_id,
                           force=False,
                           snapshots=bigger_value)
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     def test_try_set_user_quota_gigabytes_bigger_than_tenant_quota(self):
-        client = self.get_client_with_isolated_creds()
 
         # get current quotas for tenant
-        tenant_quotas = client.show_quotas(client.tenant_id)
+        tenant_quotas = self.client.show_quotas(self.tenant_id)
 
         # try set user quota for gigabytes bigger than tenant quota
         bigger_value = int(tenant_quotas["gigabytes"]) + 2
         self.assertRaises(lib_exc.BadRequest,
-                          client.update_quotas,
-                          client.tenant_id,
-                          client.user_id,
+                          self.update_quotas,
+                          self.tenant_id,
+                          user_id=self.user_id,
                           force=False,
                           gigabytes=bigger_value)
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     def test_try_set_user_quota_snap_gigabytes_bigger_than_tenant_quota(self):
-        client = self.get_client_with_isolated_creds()
-
         # get current quotas for tenant
-        tenant_quotas = client.show_quotas(client.tenant_id)
+        tenant_quotas = self.client.show_quotas(self.tenant_id)
 
         # try set user quota for snapshot gigabytes bigger than tenant quota
         bigger_value = int(tenant_quotas["snapshot_gigabytes"]) + 2
         self.assertRaises(lib_exc.BadRequest,
-                          client.update_quotas,
-                          client.tenant_id,
-                          client.user_id,
+                          self.update_quotas,
+                          self.tenant_id,
+                          user_id=self.user_id,
                           force=False,
                           snapshot_gigabytes=bigger_value)
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     def test_try_set_user_quota_share_networks_bigger_than_tenant_quota(self):
-        client = self.get_client_with_isolated_creds()
 
         # get current quotas for tenant
-        tenant_quotas = client.show_quotas(client.tenant_id)
+        tenant_quotas = self.client.show_quotas(self.tenant_id)
 
         # try set user quota for share_networks bigger than tenant quota
         bigger_value = int(tenant_quotas["share_networks"]) + 2
         self.assertRaises(lib_exc.BadRequest,
-                          client.update_quotas,
-                          client.tenant_id,
-                          client.user_id,
+                          self.update_quotas,
+                          self.tenant_id,
+                          user_id=self.user_id,
                           force=False,
                           share_networks=bigger_value)
 
@@ -219,18 +217,16 @@
     @utils.skip_if_microversion_not_supported(
         SHARE_REPLICA_QUOTAS_MICROVERSION)
     def test_try_set_user_quota_replicas_bigger_than_tenant_quota(self, key):
-        client = self.get_client_with_isolated_creds(client_version='2')
-
         # get current quotas for tenant
-        tenant_quotas = client.show_quotas(client.tenant_id)
+        tenant_quotas = self.client.show_quotas(self.tenant_id)
 
         # try set user quota for snapshots bigger than tenant quota
         bigger_value = int(tenant_quotas[key]) + 2
         kwargs = {key: bigger_value}
         self.assertRaises(lib_exc.BadRequest,
-                          client.update_quotas,
-                          client.tenant_id,
-                          client.user_id,
+                          self.update_quotas,
+                          self.tenant_id,
+                          user_id=self.user_id,
                           force=False,
                           **kwargs)
 
@@ -252,171 +248,147 @@
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     @base.skip_if_microversion_not_supported("2.7")
     def test_show_quotas_with_wrong_versions(self, url, version, method_name):
-        self.assertRaises(
-            lib_exc.NotFound,
-            getattr(self.shares_v2_client, method_name),
-            self.shares_v2_client.tenant_id,
-            version=version, url=url,
-        )
+        self.assertRaises(lib_exc.NotFound,
+                          getattr(self.client, method_name),
+                          self.tenant_id,
+                          version=version,
+                          url=url)
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     def test_show_quota_detail_with_wrong_versions(self):
         version = '2.24'
         url = 'quota-sets'
 
-        self.assertRaises(
-            lib_exc.NotFound,
-            self.shares_v2_client.detail_quotas,
-            self.shares_v2_client.tenant_id,
-            version=version, url=url,
-        )
+        self.assertRaises(lib_exc.NotFound,
+                          self.client.detail_quotas,
+                          self.tenant_id,
+                          version=version,
+                          url=url)
 
     @ddt.data('show', 'reset', 'update')
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     @base.skip_if_microversion_lt("2.39")
     def test_share_type_quotas_using_nonexistent_share_type(self, op):
-        client = self.get_client_with_isolated_creds(client_version='2')
 
         kwargs = {"share_type": "fake_nonexistent_share_type"}
         if op == 'update':
-            tenant_quotas = client.show_quotas(client.tenant_id)
+            tenant_quotas = self.client.show_quotas(self.tenant_id)
             kwargs['shares'] = tenant_quotas['shares']
 
-        self.assertRaises(
-            lib_exc.NotFound,
-            getattr(client, op + '_quotas'),
-            client.tenant_id,
-            **kwargs)
+        self.assertRaises(lib_exc.NotFound,
+                          getattr(self.client, op + '_quotas'),
+                          self.tenant_id,
+                          **kwargs)
 
     @ddt.data('id', 'name')
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     @base.skip_if_microversion_lt("2.39")
     def test_try_update_share_type_quota_for_share_networks(self, key):
-        client = self.get_client_with_isolated_creds(client_version='2')
         share_type = self._create_share_type()
-        tenant_quotas = client.show_quotas(client.tenant_id)
+        tenant_quotas = self.client.show_quotas(self.tenant_id)
 
         # Try to set 'share_networks' quota for share type
-        self.assertRaises(
-            lib_exc.BadRequest,
-            client.update_quotas,
-            client.tenant_id,
-            share_type=share_type[key],
-            share_networks=int(tenant_quotas["share_networks"]),
-        )
+        self.assertRaises(lib_exc.BadRequest,
+                          self.update_quotas,
+                          self.tenant_id,
+                          share_type=share_type[key],
+                          share_networks=int(tenant_quotas["share_networks"]))
 
     @ddt.data('share_groups', 'share_group_snapshots')
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     @base.skip_if_microversion_lt(SHARE_GROUPS_MICROVERSION)
     def test_try_update_share_type_quota_for_share_groups(self, quota_name):
-        client = self.get_client_with_isolated_creds(client_version='2')
         share_type = self._create_share_type()
-        tenant_quotas = client.show_quotas(client.tenant_id)
+        tenant_quotas = self.client.show_quotas(self.tenant_id)
 
-        self.assertRaises(
-            lib_exc.BadRequest,
-            client.update_quotas,
-            client.tenant_id,
-            share_type=share_type["name"],
-            **{quota_name: int(tenant_quotas[quota_name])}
-        )
+        self.assertRaises(lib_exc.BadRequest,
+                          self.update_quotas,
+                          self.tenant_id,
+                          share_type=share_type["name"],
+                          **{quota_name: int(tenant_quotas[quota_name])})
 
     @ddt.data('share_groups', 'share_group_snapshots')
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     @base.skip_if_microversion_not_supported(PRE_SHARE_GROUPS_MICROVERSION)
     @base.skip_if_microversion_not_supported(SHARE_GROUPS_MICROVERSION)
     def test_share_group_quotas_using_too_old_microversion(self, quota_key):
-        client = self.get_client_with_isolated_creds(client_version='2')
-        tenant_quotas = client.show_quotas(
-            client.tenant_id, version=SHARE_GROUPS_MICROVERSION)
+        tenant_quotas = self.client.show_quotas(
+            self.tenant_id, version=SHARE_GROUPS_MICROVERSION)
         kwargs = {
             "version": PRE_SHARE_GROUPS_MICROVERSION,
             quota_key: tenant_quotas[quota_key],
         }
 
-        self.assertRaises(
-            lib_exc.BadRequest,
-            client.update_quotas,
-            client.tenant_id,
-            **kwargs)
+        self.assertRaises(lib_exc.BadRequest,
+                          self.client.update_quotas,
+                          self.tenant_id,
+                          **kwargs)
 
     @ddt.data("share_replicas", "replica_gigabytes")
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     @base.skip_if_microversion_not_supported(SHARE_REPLICA_QUOTAS_MICROVERSION)
     def test_share_replica_quotas_using_too_old_microversion(self, quota_key):
-        client = self.get_client_with_isolated_creds(client_version='2')
-        tenant_quotas = client.show_quotas(
-            client.tenant_id, version=SHARE_REPLICA_QUOTAS_MICROVERSION)
+        tenant_quotas = self.client.show_quotas(
+            self.tenant_id, version=SHARE_REPLICA_QUOTAS_MICROVERSION)
         kwargs = {
             "version": PRE_SHARE_REPLICA_QUOTAS_MICROVERSION,
             quota_key: tenant_quotas[quota_key],
         }
 
-        self.assertRaises(
-            lib_exc.BadRequest,
-            client.update_quotas,
-            client.tenant_id,
-            **kwargs)
+        self.assertRaises(lib_exc.BadRequest,
+                          self.update_quotas,
+                          self.tenant_id,
+                          **kwargs)
 
     @ddt.data('show', 'reset', 'update')
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     @base.skip_if_microversion_lt("2.38")
     def test_share_type_quotas_using_too_old_microversion(self, op):
-        client = self.get_client_with_isolated_creds(client_version='2')
         share_type = self._create_share_type()
         kwargs = {"version": "2.38", "share_type": share_type["name"]}
         if op == 'update':
-            tenant_quotas = client.show_quotas(client.tenant_id)
+            tenant_quotas = self.client.show_quotas(self.tenant_id)
             kwargs['shares'] = tenant_quotas['shares']
 
-        self.assertRaises(
-            lib_exc.BadRequest,
-            getattr(client, op + '_quotas'),
-            client.tenant_id,
-            **kwargs)
+        self.assertRaises(lib_exc.BadRequest,
+                          getattr(self.client, op + '_quotas'),
+                          self.tenant_id,
+                          **kwargs)
 
     @ddt.data('show', 'reset', 'update')
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     @base.skip_if_microversion_lt("2.39")
     def test_quotas_providing_share_type_and_user_id(self, op):
-        client = self.get_client_with_isolated_creds(client_version='2')
         share_type = self._create_share_type()
-        kwargs = {"share_type": share_type["name"], "user_id": client.user_id}
+        kwargs = {"share_type": share_type["name"], "user_id": self.user_id}
         if op == 'update':
-            tenant_quotas = client.show_quotas(client.tenant_id)
+            tenant_quotas = self.client.show_quotas(self.tenant_id)
             kwargs['shares'] = tenant_quotas['shares']
 
-        self.assertRaises(
-            lib_exc.BadRequest,
-            getattr(client, op + '_quotas'),
-            client.tenant_id,
-            **kwargs)
+        self.assertRaises(lib_exc.BadRequest,
+                          getattr(self.client, op + '_quotas'),
+                          self.tenant_id,
+                          **kwargs)
 
     @ddt.data(11, -1)
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     @base.skip_if_microversion_lt("2.39")
     def test_update_share_type_quotas_bigger_than_project_quota(self, st_q):
-        client = self.get_client_with_isolated_creds(client_version='2')
         share_type = self._create_share_type()
-        client.update_quotas(client.tenant_id, shares=10)
+        self.update_quotas(self.tenant_id, shares=10)
 
-        self.assertRaises(
-            lib_exc.BadRequest,
-            client.update_quotas,
-            client.tenant_id,
-            share_type=share_type['name'],
-            force=False,
-            shares=st_q)
+        self.assertRaises(lib_exc.BadRequest,
+                          self.update_quotas,
+                          self.tenant_id,
+                          share_type=share_type['name'],
+                          force=False,
+                          shares=st_q)
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
     def test_create_share_over_quota_limit(self):
-        original_quota = self.shares_v2_client.show_quotas(
-            self.tenant_id)
         self.create_share(share_type_id=self.share_type_id)
-        self.shares_v2_client.update_quotas(self.tenant_id, shares=1)
-        updated_quota = self.shares_v2_client.show_quotas(self.tenant_id)
-        self.addCleanup(self.shares_v2_client.update_quotas,
-                        self.tenant_id, shares=original_quota['shares'])
+
+        updated_quota = self.update_quotas(self.tenant_id, shares=1)
         self.assertEqual(1, updated_quota['shares'])
 
         self.assertRaises(lib_exc.OverLimit,
@@ -427,32 +399,29 @@
 @ddt.ddt
 class ReplicaQuotasNegativeTest(rep_neg_test.ReplicationNegativeBase):
 
+    # We want to force some fresh projects for this test class, since we'll be
+    # manipulating project quotas - and any pre-existing projects may have
+    # resources, quotas and the like that might interfere with our test cases.
+    force_tenant_isolation = True
+
     @classmethod
     def skip_checks(cls):
         super(ReplicaQuotasNegativeTest, cls).skip_checks()
+        if not CONF.auth.use_dynamic_credentials:
+            raise cls.skipException('Dynamic credentials are required')
         if not CONF.share.run_quota_tests:
             msg = "Quota tests are disabled."
             raise cls.skipException(msg)
-
         utils.check_skip_if_microversion_lt(SHARE_REPLICA_QUOTAS_MICROVERSION)
 
     def _modify_quotas_for_test(self, quota_key, new_limit):
         kwargs = {quota_key: new_limit}
 
-        # Get the original quota values
-        original_quota = self.admin_client.show_quotas(self.tenant_id)
-
         # Update the current quotas
-        self.admin_client.update_quotas(self.tenant_id, **kwargs)
-
-        # Save the previous value
-        old_quota_values = {quota_key: original_quota[quota_key]}
+        self.update_quotas(self.tenant_id, client=self.admin_client, **kwargs)
 
         # Get the updated quotas and add a cleanup
         updated_quota = self.admin_client.show_quotas(self.tenant_id)
-        self.addCleanup(self.admin_client.update_quotas,
-                        self.tenant_id,
-                        **old_quota_values)
 
         # Make sure that the new value was properly set
         self.assertEqual(new_limit, updated_quota[quota_key])
@@ -493,6 +462,6 @@
 
         # Make sure that the request to create a third one will fail
         self.assertRaises(lib_exc.OverLimit,
-                          self.shares_v2_client.extend_share,
+                          self.client.extend_share,
                           self.share1['id'],
                           new_size)
diff --git a/manila_tempest_tests/tests/api/admin/test_replication.py b/manila_tempest_tests/tests/api/admin/test_replication.py
index 60f97a7..9e6a323 100644
--- a/manila_tempest_tests/tests/api/admin/test_replication.py
+++ b/manila_tempest_tests/tests/api/admin/test_replication.py
@@ -50,7 +50,7 @@
             )
 
         extra_specs = {"replication_type": cls.replication_type}
-        cls.share_type = cls._create_share_type(extra_specs)
+        cls.share_type = cls._create_share_type(specs=extra_specs)
         cls.share_type_id = cls.share_type['id']
         cls.sn_id = None
         if cls.multitenancy_enabled:
diff --git a/manila_tempest_tests/tests/api/admin/test_replication_actions.py b/manila_tempest_tests/tests/api/admin/test_replication_actions.py
index 8264798..428229c 100644
--- a/manila_tempest_tests/tests/api/admin/test_replication_actions.py
+++ b/manila_tempest_tests/tests/api/admin/test_replication_actions.py
@@ -53,7 +53,7 @@
 
         # create share type
         extra_specs = {"replication_type": cls.replication_type}
-        cls.share_type = cls._create_share_type(extra_specs)
+        cls.share_type = cls._create_share_type(specs=extra_specs)
         cls.share_type_id = cls.share_type['id']
 
         cls.sn_id = None
diff --git a/manila_tempest_tests/tests/api/admin/test_share_networks.py b/manila_tempest_tests/tests/api/admin/test_share_networks.py
index e66e503..b4211bd 100644
--- a/manila_tempest_tests/tests/api/admin/test_share_networks.py
+++ b/manila_tempest_tests/tests/api/admin/test_share_networks.py
@@ -19,9 +19,8 @@
 from manila_tempest_tests.tests.api import test_share_networks
 
 
-class ShareNetworkAdminTest(
-        base.BaseSharesAdminTest,
-        test_share_networks.ShareNetworkListMixin):
+class ShareNetworkAdminTest(base.BaseSharesMixedTest,
+                            test_share_networks.ShareNetworkListMixin):
 
     @classmethod
     def resource_setup(cls):
@@ -49,8 +48,6 @@
             cls.sn_with_ldap_ss["id"],
             cls.ss_ldap["id"])
 
-        cls.isolated_client = cls.get_client_with_isolated_creds(
-            type_of_creds='alt')
         cls.data_sn_with_kerberos_ss = {
             'name': 'sn_with_kerberos_ss',
             'created_at': '2003-03-03',
@@ -64,21 +61,22 @@
             'description': 'fake description',
         }
 
-        cls.ss_kerberos = cls.isolated_client.create_security_service(
+        cls.ss_kerberos = cls.alt_shares_v2_client.create_security_service(
             ss_type='kerberos',
             **cls.data_sn_with_ldap_ss)
 
-        cls.sn_with_kerberos_ss = cls.isolated_client.create_share_network(
-            cleanup_in_class=True,
-            **cls.data_sn_with_kerberos_ss)
+        cls.sn_with_kerberos_ss = (
+            cls.alt_shares_v2_client.create_share_network(
+                cleanup_in_class=True, **cls.data_sn_with_kerberos_ss)
+        )
 
-        cls.isolated_client.add_sec_service_to_share_network(
+        cls.alt_shares_v2_client.add_sec_service_to_share_network(
             cls.sn_with_kerberos_ss["id"],
             cls.ss_kerberos["id"])
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_list_share_networks_all_tenants(self):
-        listed = self.shares_client.list_share_networks_with_detail(
+        listed = self.admin_shares_v2_client.list_share_networks_with_detail(
             {'all_tenants': 1})
         self.assertTrue(any(self.sn_with_ldap_ss['id'] == sn['id']
                             for sn in listed))
@@ -87,7 +85,7 @@
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_API)
     def test_list_share_networks_filter_by_project_id(self):
-        listed = self.shares_client.list_share_networks_with_detail(
+        listed = self.admin_shares_v2_client.list_share_networks_with_detail(
             {'project_id': self.sn_with_kerberos_ss['project_id']})
         self.assertTrue(any(self.sn_with_kerberos_ss['id'] == sn['id']
                             for sn in listed))
diff --git a/manila_tempest_tests/tests/api/admin/test_shares_actions.py b/manila_tempest_tests/tests/api/admin/test_shares_actions.py
index dfadd7a..f4ccfa4 100644
--- a/manila_tempest_tests/tests/api/admin/test_shares_actions.py
+++ b/manila_tempest_tests/tests/api/admin/test_shares_actions.py
@@ -36,7 +36,7 @@
 
         # create share type for share filtering purposes
         specs = {"storage_protocol": CONF.share.capability_storage_protocol}
-        cls.share_type = cls._create_share_type(specs)
+        cls.share_type = cls._create_share_type(specs=specs)
         cls.share_type_id = cls.share_type['id']
 
         # create share
diff --git a/manila_tempest_tests/tests/api/admin/test_snapshot_export_locations_negative.py b/manila_tempest_tests/tests/api/admin/test_snapshot_export_locations_negative.py
index 9d48254..10d6193 100644
--- a/manila_tempest_tests/tests/api/admin/test_snapshot_export_locations_negative.py
+++ b/manila_tempest_tests/tests/api/admin/test_snapshot_export_locations_negative.py
@@ -39,7 +39,7 @@
     def setup_clients(cls):
         super(SnapshotExportLocationsNegativeTest, cls).setup_clients()
         cls.admin_client = cls.admin_shares_v2_client
-        cls.isolated_client = cls.alt_shares_v2_client
+        cls.different_project_client = cls.alt_shares_v2_client
 
     @classmethod
     def resource_setup(cls):
@@ -66,15 +66,15 @@
         )
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
-    def test_list_snapshot_export_locations_by_member(self):
+    def test_list_snapshot_export_locations_by_different_project_user(self):
         self.assertRaises(
             lib_exc.NotFound,
-            self.isolated_client.list_snapshot_export_locations,
+            self.different_project_client.list_snapshot_export_locations,
             self.snapshot['id']
         )
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
-    def test_get_snapshot_export_location_by_member(self):
+    def test_get_snapshot_export_location_by_different_project_user(self):
         export_locations = (
             self.admin_client.list_snapshot_export_locations(
                 self.snapshot['id']))
@@ -84,10 +84,9 @@
                 continue
             self.assertRaises(
                 lib_exc.NotFound,
-                self.isolated_client.get_snapshot_export_location,
+                self.different_project_client.get_snapshot_export_location,
                 self.snapshot['id'],
-                export_location['id']
-            )
+                export_location['id'])
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
     def test_get_inexistent_snapshot_instance_export_location(self):
@@ -108,7 +107,8 @@
             for el in export_locations:
                 self.assertRaises(
                     lib_exc.Forbidden,
-                    self.isolated_client.get_snapshot_instance_export_location,
+                    (self.different_project_client.
+                     get_snapshot_instance_export_location),
                     snapshot_instance['id'], el['id'],
                 )
 
@@ -129,7 +129,9 @@
     def setup_clients(cls):
         super(SnapshotExportLocationsAPIOnlyNegativeTest, cls).setup_clients()
         cls.admin_client = cls.admin_shares_v2_client
-        cls.isolated_client = cls.alt_shares_v2_client
+        # admin_member_client is a regular user in admin's project
+        cls.admin_member_client = (
+            cls.admin_project_member_client.shares_v2_client)
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     def test_list_export_locations_by_nonexistent_snapshot(self):
@@ -152,6 +154,6 @@
             self):
         self.assertRaises(
             lib_exc.Forbidden,
-            self.isolated_client.list_snapshot_instance_export_locations,
+            self.admin_member_client.list_snapshot_instance_export_locations,
             "fake-inexistent-snapshot-instance-id"
         )
diff --git a/manila_tempest_tests/tests/api/admin/test_user_messages_negative.py b/manila_tempest_tests/tests/api/admin/test_user_messages_negative.py
index cf592e4..7d77eec 100644
--- a/manila_tempest_tests/tests/api/admin/test_user_messages_negative.py
+++ b/manila_tempest_tests/tests/api/admin/test_user_messages_negative.py
@@ -24,7 +24,7 @@
 QUERY_BY_TIMESTAMP_MICROVERSION = '2.52'
 
 
-class UserMessageNegativeTest(base.BaseSharesAdminTest):
+class UserMessageNegativeTest(base.BaseSharesMixedTest):
 
     @classmethod
     def skip_checks(cls):
@@ -37,10 +37,8 @@
 
     @decorators.attr(type=[base.TAG_NEGATIVE, base.TAG_API])
     def test_show_message_of_other_tenants(self):
-        isolated_client = self.get_client_with_isolated_creds(
-            type_of_creds='alt', client_version='2')
         self.assertRaises(lib_exc.NotFound,
-                          isolated_client.get_message,
+                          self.alt_shares_v2_client.get_message,
                           self.message['id'])
 
     @decorators.attr(type=[base.TAG_NEGATIVE, base.TAG_API])
@@ -51,10 +49,8 @@
 
     @decorators.attr(type=[base.TAG_NEGATIVE, base.TAG_API])
     def test_delete_message_of_other_tenants(self):
-        isolated_client = self.get_client_with_isolated_creds(
-            type_of_creds='alt', client_version='2')
         self.assertRaises(lib_exc.NotFound,
-                          isolated_client.delete_message,
+                          self.alt_shares_v2_client.delete_message,
                           self.message['id'])
 
     @decorators.attr(type=[base.TAG_NEGATIVE, base.TAG_API])
diff --git a/manila_tempest_tests/tests/api/base.py b/manila_tempest_tests/tests/api/base.py
index 2ed43c7..fc5c6a1 100644
--- a/manila_tempest_tests/tests/api/base.py
+++ b/manila_tempest_tests/tests/api/base.py
@@ -14,7 +14,6 @@
 #    under the License.
 
 import copy
-import inspect
 import re
 import traceback
 
@@ -100,12 +99,6 @@
 def network_synchronized(f):
 
     def wrapped_func(self, *args, **kwargs):
-        with_isolated_creds = True if len(args) > 2 else False
-        no_lock_required = kwargs.get(
-            "isolated_creds_client", with_isolated_creds)
-        if no_lock_required:
-            # Usage of not reusable network. No need in lock.
-            return f(self, *args, **kwargs)
 
         # Use lock assuming reusage of common network.
         @lockutils.synchronized("manila_network_lock", external=True)
@@ -134,12 +127,6 @@
     # Will be cleaned up in tearDown method
     method_resources = []
 
-    # Will be cleaned up in resource_cleanup
-    class_isolated_creds = []
-
-    # Will be cleaned up in tearDown method
-    method_isolated_creds = []
-
     # NOTE(andreaf) Override the client manager class to be used, so that
     # a stable class is used, which includes plugin registered services as well
     client_manager = clients.Clients
@@ -187,73 +174,6 @@
             identity_uri=identity_uri)
 
     @classmethod
-    def get_client_with_isolated_creds(cls,
-                                       name=None,
-                                       type_of_creds="admin",
-                                       cleanup_in_class=False,
-                                       client_version='1'):
-        """Creates isolated creds.
-
-        :param name: name, will be used for naming ic and related stuff
-        :param type_of_creds: admin, alt or primary
-        :param cleanup_in_class: defines place where to delete
-        :returns: SharesClient -- shares client with isolated creds.
-        :returns: To client added dict attr 'creds' with
-        :returns: key elements 'tenant' and 'user'.
-        """
-        if name is None:
-            # Get name of test method
-            name = inspect.stack()[1][3]
-            if len(name) > 32:
-                name = name[0:32]
-
-        # Choose type of isolated creds
-        ic = cls._get_dynamic_creds(name)
-        if "admin" in type_of_creds:
-            creds = ic.get_admin_creds().credentials
-        elif "alt" in type_of_creds:
-            creds = ic.get_alt_creds().credentials
-        else:
-            creds = ic.get_credentials(type_of_creds).credentials
-        ic.type_of_creds = type_of_creds
-
-        # create client with isolated creds
-        os = clients.Clients(creds)
-        if client_version == '1':
-            client = os.share_v1.SharesClient()
-        elif client_version == '2':
-            client = os.share_v2.SharesV2Client()
-
-        # Set place where will be deleted isolated creds
-        ic_res = {
-            "method": ic.clear_creds,
-            "deleted": False,
-        }
-        if cleanup_in_class:
-            cls.class_isolated_creds.insert(0, ic_res)
-        else:
-            cls.method_isolated_creds.insert(0, ic_res)
-
-        # Provide share network
-        if CONF.share.multitenancy_enabled:
-            if (not CONF.service_available.neutron and
-                    CONF.share.create_networks_when_multitenancy_enabled):
-                raise cls.skipException("Neutron support is required")
-            nc = os.network.NetworksClient()
-            share_network_id = cls.provide_share_network(client, nc, ic)
-            client.share_network_id = share_network_id
-            resource = {
-                "type": "share_network",
-                "id": client.share_network_id,
-                "client": client,
-            }
-            if cleanup_in_class:
-                cls.class_resources.insert(0, resource)
-            else:
-                cls.method_resources.insert(0, resource)
-        return client
-
-    @classmethod
     def skip_checks(cls):
         super(BaseSharesTest, cls).skip_checks()
         if not CONF.service_available.manila:
@@ -307,20 +227,17 @@
 
     def setUp(self):
         super(BaseSharesTest, self).setUp()
-        self.addCleanup(self.clear_isolated_creds)
         self.addCleanup(self.clear_resources)
         verify_test_has_appropriate_tags(self)
 
     @classmethod
     def resource_cleanup(cls):
         cls.clear_resources(cls.class_resources)
-        cls.clear_isolated_creds(cls.class_isolated_creds)
         super(BaseSharesTest, cls).resource_cleanup()
 
     @classmethod
     @network_synchronized
     def provide_share_network(cls, shares_client, networks_client,
-                              isolated_creds_client=None,
                               ignore_multitenancy_config=False):
         """Used for finding/creating share network for multitenant driver.
 
@@ -329,9 +246,6 @@
 
         :param shares_client: shares client, which requires share-network
         :param networks_client: network client from same tenant as shares
-        :param isolated_creds_client: DynamicCredentialProvider instance
-            If provided, then its networking will be used if needed.
-            If not provided, then common network will be used if needed.
         :param ignore_multitenancy_config: provide a share network regardless
             of 'multitenancy_enabled' configuration value.
         :returns: str -- share network id for shares_client tenant
@@ -375,50 +289,41 @@
                     share_network_id = sn["id"]
             else:
                 net_id = subnet_id = share_network_id = None
-
-                if not isolated_creds_client:
-                    # Search for networks, created in previous runs
-                    service_net_name = "share-service"
-                    networks = networks_client.list_networks()
-                    if "networks" in networks.keys():
-                        networks = networks["networks"]
-                    for network in networks:
-                        if (service_net_name in network["name"] and
-                                sc.tenant_id == network['tenant_id']):
-                            net_id = network["id"]
-                            if len(network["subnets"]) > 0:
-                                subnet_id = network["subnets"][0]
-                                break
-
-                    # Create suitable network
-                    if net_id is None or subnet_id is None:
-                        ic = cls._get_dynamic_creds(service_net_name)
-                        net_data = ic._create_network_resources(sc.tenant_id)
-                        network, subnet, router = net_data
+                # Search for networks, created in previous runs
+                service_net_name = "share-service"
+                networks = networks_client.list_networks()
+                if "networks" in networks.keys():
+                    networks = networks["networks"]
+                for network in networks:
+                    if (service_net_name in network["name"] and
+                            sc.tenant_id == network['tenant_id']):
                         net_id = network["id"]
-                        subnet_id = subnet["id"]
-
-                    # Try get suitable share-network
-                    share_networks = sc.list_share_networks_with_detail()
-                    for sn in share_networks:
-                        net_info = (
-                            utils.share_network_get_default_subnet(sn)
-                            if utils.share_network_subnets_are_supported()
-                            else sn)
-                        if net_info is None:
-                            continue
-                        if (net_id == net_info["neutron_net_id"] and
-                                subnet_id == net_info["neutron_subnet_id"] and
-                                sn["name"] and search_word in sn["name"]):
-                            share_network_id = sn["id"]
+                        if len(network["subnets"]) > 0:
+                            subnet_id = network["subnets"][0]
                             break
-                else:
-                    sn_name = "autogenerated_by_tempest_for_isolated_creds"
-                    # Use precreated network and subnet from isolated creds
-                    net_id = isolated_creds_client.get_credentials(
-                        isolated_creds_client.type_of_creds).network['id']
-                    subnet_id = isolated_creds_client.get_credentials(
-                        isolated_creds_client.type_of_creds).subnet['id']
+
+                # Create suitable network
+                if net_id is None or subnet_id is None:
+                    ic = cls._get_dynamic_creds(service_net_name)
+                    net_data = ic._create_network_resources(sc.tenant_id)
+                    network, subnet, router = net_data
+                    net_id = network["id"]
+                    subnet_id = subnet["id"]
+
+                # Try get suitable share-network
+                share_networks = sc.list_share_networks_with_detail()
+                for sn in share_networks:
+                    net_info = (
+                        utils.share_network_get_default_subnet(sn)
+                        if utils.share_network_subnets_are_supported()
+                        else sn)
+                    if net_info is None:
+                        continue
+                    if (net_id == net_info["neutron_net_id"] and
+                            subnet_id == net_info["neutron_subnet_id"] and
+                            sn["name"] and search_word in sn["name"]):
+                        share_network_id = sn["id"]
+                        break
 
                 # Create suitable share-network
                 if share_network_id is None:
@@ -799,14 +704,15 @@
             status_attr="replica_state")
         return replica
 
-    def _get_access_rule_data_from_config(self):
+    @classmethod
+    def _get_access_rule_data_from_config(cls):
         """Get the first available access type/to combination from config.
 
         This method opportunistically picks the first configured protocol
         to create the share. Do not use this method in tests where you need
         to test depth and breadth in the access types and access recipients.
         """
-        protocol = self.shares_v2_client.share_protocol
+        protocol = cls.shares_v2_client.share_protocol
 
         if protocol in CONF.share.enable_ip_rules_for_protocols:
             access_type = "ip"
@@ -822,7 +728,7 @@
             access_to = "eve"
         else:
             message = "Unrecognized protocol and access rules configuration."
-            raise self.skipException(message)
+            raise cls.skipException(message)
 
         return access_type, access_to
 
@@ -907,6 +813,23 @@
                                               is_public, description)
         return share_type
 
+    @classmethod
+    def update_quotas(cls, project_id, user_id=None, cleanup=True,
+                      client=None, **kwargs):
+        client = client or cls.shares_v2_client
+        updated_quotas = client.update_quotas(project_id,
+                                              user_id=user_id,
+                                              **kwargs)
+        resource = {
+            "type": "quotas",
+            "id": project_id,
+            "client": client,
+            "user_id": user_id,
+        }
+        if cleanup:
+            cls.method_resources.insert(0, resource)
+        return updated_quotas
+
     @staticmethod
     def add_extra_specs_to_dict(extra_specs=None):
         """Add any required extra-specs to share type dictionary"""
@@ -934,18 +857,6 @@
         return extra_specs_dict
 
     @classmethod
-    def clear_isolated_creds(cls, creds=None):
-        if creds is None:
-            creds = cls.method_isolated_creds
-        for ic in creds:
-            if "deleted" not in ic.keys():
-                ic["deleted"] = False
-            if not ic["deleted"]:
-                with handle_cleanup_exceptions():
-                    ic["method"]()
-                ic["deleted"] = True
-
-    @classmethod
     def clear_share_replicas(cls, share_id, client=None):
         client = client or cls.shares_v2_client
         share_replicas = client.list_share_replicas(
@@ -1023,6 +934,9 @@
                         client.wait_for_resource_deletion(
                             share_network_subnet_id=res_id,
                             sn_id=sn_id)
+                    elif res["type"] == "quotas":
+                        user_id = res.get('user_id')
+                        client.reset_quotas(res_id, user_id=user_id)
                     else:
                         LOG.warning("Provided unsupported resource type for "
                                     "cleanup '%s'. Skipping.", res["type"])
@@ -1129,6 +1043,7 @@
         share_type_name = data_utils.rand_name("share-type")
 
         bogus_type = self.create_share_type(
+            client=self.admin_shares_v2_client,
             name=share_type_name,
             extra_specs=extra_specs)['share_type']
 
@@ -1156,11 +1071,11 @@
         cls.admin_shares_v2_client = cls.os_admin.share_v2.SharesV2Client()
 
     @classmethod
-    def _create_share_type(cls, specs=None):
+    def _create_share_type(cls, is_public=True, specs=None):
         name = data_utils.rand_name("unique_st_name")
         extra_specs = cls.add_extra_specs_to_dict(specs)
         return cls.create_share_type(
-            name, extra_specs=extra_specs,
+            name, extra_specs=extra_specs, is_public=is_public,
             client=cls.admin_shares_v2_client)['share_type']
 
     @classmethod
@@ -1354,11 +1269,11 @@
         return os
 
     @classmethod
-    def _create_share_type(cls, specs=None):
+    def _create_share_type(cls, is_public=True, specs=None):
         name = data_utils.rand_name("unique_st_name")
         extra_specs = cls.add_extra_specs_to_dict(specs)
         return cls.create_share_type(
-            name, extra_specs=extra_specs,
+            name, extra_specs=extra_specs, is_public=is_public,
             client=cls.admin_shares_v2_client)['share_type']
 
     @classmethod
diff --git a/manila_tempest_tests/tests/api/test_access_rules_metadata.py b/manila_tempest_tests/tests/api/test_access_rules_metadata.py
index e569585..1c2db41 100644
--- a/manila_tempest_tests/tests/api/test_access_rules_metadata.py
+++ b/manila_tempest_tests/tests/api/test_access_rules_metadata.py
@@ -26,6 +26,12 @@
 
 @ddt.ddt
 class AccessRulesMetadataTest(base.BaseSharesMixedTest):
+    """A Test class to test access rule metadata generically.
+
+    Tests in this class don't care about the type of access rule or the
+    protocol of the share created. They are meant to test the API semantics
+    of the access rule metadata APIs.
+    """
 
     @classmethod
     def skip_checks(cls):
@@ -47,22 +53,8 @@
     @classmethod
     def resource_setup(cls):
         super(AccessRulesMetadataTest, cls).resource_setup()
-        # The share access rule metadata doesn't care about the value of
-        # access type, access protocol, access_to, so we only get one of
-        # the value that the driver support.
-        if CONF.share.enable_ip_rules_for_protocols:
-            cls.protocol = CONF.share.enable_ip_rules_for_protocols[0]
-            cls.access_type = "ip"
-        elif CONF.share.enable_user_rules_for_protocols:
-            cls.protocol = CONF.share.enable_user_rules_for_protocols[0]
-            cls.access_type = "user"
-        elif CONF.share.enable_cert_rules_for_protocols:
-            cls.protocol = CONF.share.enable_cert_rules_for_protocols[0]
-            cls.access_type = "cert"
-        elif CONF.share.enable_cephx_rules_for_protocols:
-            cls.protocol = CONF.share.enable_cephx_rules_for_protocols[0]
-            cls.access_type = "cephx"
-        cls.shares_v2_client.share_protocol = cls.protocol
+        cls.protocol = cls.shares_v2_client.share_protocol
+        cls.access_type, __ = cls._get_access_rule_data_from_config()
         int_range = range(20, 50)
         cls.access_to = {
             # list of unique values is required for ability to create lots
diff --git a/manila_tempest_tests/tests/api/test_access_rules_metadata_negative.py b/manila_tempest_tests/tests/api/test_access_rules_metadata_negative.py
index 033beea..4383a84 100644
--- a/manila_tempest_tests/tests/api/test_access_rules_metadata_negative.py
+++ b/manila_tempest_tests/tests/api/test_access_rules_metadata_negative.py
@@ -27,6 +27,12 @@
 
 @ddt.ddt
 class AccessesMetadataNegativeTest(base.BaseSharesMixedTest):
+    """A Test class with generic negative access rule metadata tests.
+
+    Tests in this class don't care about the type of access rule or the
+    protocol of the share created. They are meant to test the API semantics
+    of the access rule metadata APIs.
+    """
 
     @classmethod
     def skip_checks(cls):
@@ -48,23 +54,10 @@
     @classmethod
     def resource_setup(cls):
         super(AccessesMetadataNegativeTest, cls).resource_setup()
-        if CONF.share.enable_ip_rules_for_protocols:
-            cls.protocol = CONF.share.enable_ip_rules_for_protocols[0]
-            cls.access_type = "ip"
-            cls.access_to = utils.rand_ip()
-        elif CONF.share.enable_user_rules_for_protocols:
-            cls.protocol = CONF.share.enable_user_rules_for_protocols[0]
-            cls.access_type = "user"
-            cls.access_to = CONF.share.username_for_user_rules
-        elif CONF.share.enable_cert_rules_for_protocols:
-            cls.protocol = CONF.share.enable_cert_rules_for_protocols[0]
-            cls.access_type = "cert"
-            cls.access_to = "client3.com"
-        elif CONF.share.enable_cephx_rules_for_protocols:
-            cls.protocol = CONF.share.enable_cephx_rules_for_protocols[0]
-            cls.access_type = "cephx"
-            cls.access_to = "eve"
-        cls.shares_v2_client.share_protocol = cls.protocol
+        cls.protocol = cls.shares_v2_client.share_protocol
+        cls.access_type, cls.access_to = (
+            cls._get_access_rule_data_from_config()
+        )
         # create share type
         cls.share_type = cls._create_share_type()
         cls.share_type_id = cls.share_type['id']
diff --git a/manila_tempest_tests/tests/api/test_replication_negative.py b/manila_tempest_tests/tests/api/test_replication_negative.py
index 3215641..2d66d38 100644
--- a/manila_tempest_tests/tests/api/test_replication_negative.py
+++ b/manila_tempest_tests/tests/api/test_replication_negative.py
@@ -52,7 +52,7 @@
 
         # create share type
         extra_specs = {"replication_type": cls.replication_type}
-        cls.share_type = cls._create_share_type(extra_specs)
+        cls.share_type = cls._create_share_type(specs=extra_specs)
         cls.share_type_id = cls.share_type['id']
         cls.sn_id = None
         if cls.multitenancy_enabled:
diff --git a/manila_tempest_tests/tests/api/test_replication_snapshots.py b/manila_tempest_tests/tests/api/test_replication_snapshots.py
index 591b3fd..605275c 100644
--- a/manila_tempest_tests/tests/api/test_replication_snapshots.py
+++ b/manila_tempest_tests/tests/api/test_replication_snapshots.py
@@ -53,7 +53,7 @@
 
         # create share type
         extra_specs = {"replication_type": cls.replication_type}
-        cls.share_type = cls._create_share_type(extra_specs)
+        cls.share_type = cls._create_share_type(specs=extra_specs)
         cls.share_type_id = cls.share_type['id']
         cls.sn_id = None
         if cls.multitenancy_enabled:
diff --git a/manila_tempest_tests/tests/api/test_rules.py b/manila_tempest_tests/tests/api/test_rules.py
index 63e9c3e..42d654e 100644
--- a/manila_tempest_tests/tests/api/test_rules.py
+++ b/manila_tempest_tests/tests/api/test_rules.py
@@ -523,6 +523,12 @@
 
 @ddt.ddt
 class ShareRulesTest(base.BaseSharesMixedTest):
+    """A Test class to test access rules generically.
+
+    Tests in this class don't care about the type of access rule or the
+    protocol of the share created. They are meant to test the API semantics
+    of the access rules APIs.
+    """
 
     @classmethod
     def skip_checks(cls):
@@ -541,23 +547,10 @@
     @classmethod
     def resource_setup(cls):
         super(ShareRulesTest, cls).resource_setup()
-        if CONF.share.enable_ip_rules_for_protocols:
-            cls.protocol = CONF.share.enable_ip_rules_for_protocols[0]
-            cls.access_type = "ip"
-            cls.access_to = "8.8.8.8"
-        elif CONF.share.enable_user_rules_for_protocols:
-            cls.protocol = CONF.share.enable_user_rules_for_protocols[0]
-            cls.access_type = "user"
-            cls.access_to = CONF.share.username_for_user_rules
-        elif CONF.share.enable_cert_rules_for_protocols:
-            cls.protocol = CONF.share.enable_cert_rules_for_protocols[0]
-            cls.access_type = "cert"
-            cls.access_to = "client3.com"
-        elif CONF.share.enable_cephx_rules_for_protocols:
-            cls.protocol = CONF.share.enable_cephx_rules_for_protocols[0]
-            cls.access_type = "cephx"
-            cls.access_to = "eve"
-        cls.shares_v2_client.share_protocol = cls.protocol
+        cls.protocol = cls.shares_v2_client.share_protocol
+        cls.access_type, cls.access_to = (
+            cls._get_access_rule_data_from_config()
+        )
         cls.share_type = cls._create_share_type()
         cls.share_type_id = cls.share_type['id']
         cls.share = cls.create_share(share_type_id=cls.share_type_id)
diff --git a/manila_tempest_tests/tests/api/test_rules_negative.py b/manila_tempest_tests/tests/api/test_rules_negative.py
index 764533c..0fb629e 100644
--- a/manila_tempest_tests/tests/api/test_rules_negative.py
+++ b/manila_tempest_tests/tests/api/test_rules_negative.py
@@ -20,6 +20,7 @@
 from testtools import testcase as tc
 
 from manila_tempest_tests.common import constants
+from manila_tempest_tests import share_exceptions
 from manila_tempest_tests.tests.api import base
 from manila_tempest_tests import utils
 
@@ -385,17 +386,27 @@
                           self.share["id"], self.access_type, self.access_to,
                           access_level="su")
 
+    @tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
+    def test_different_tenants_cannot_use_same_cephx_id(self):
+        # Grant access to the share
+        access1 = self.shares_v2_client.create_access_rule(
+            self.share['id'], self.access_type, self.access_to, 'rw')
+        self.shares_v2_client.wait_for_access_rule_status(
+            self.share['id'], access1['id'], 'active')
 
-def skip_if_cephx_access_type_not_supported_by_client(self, client):
-    if client == 'shares_client':
-        version = '1.0'
-    else:
-        version = LATEST_MICROVERSION
-    if (CONF.share.enable_cephx_rules_for_protocols and
-            utils.is_microversion_lt(version, '2.13')):
-        msg = ("API version %s does not support cephx access type, need "
-               "version >= 2.13." % version)
-        raise self.skipException(msg)
+        # Create second share by the new user
+        share2 = self.create_share(client=self.alt_shares_v2_client,
+                                   share_protocol=self.protocol,
+                                   share_type_id=self.share_type_id)
+
+        # Try grant access to the second share using the same cephx id as used
+        # on the first share
+        access2 = self.alt_shares_v2_client.create_access_rule(
+            share2['id'], self.access_type, self.access_to, 'rw')
+        self.assertRaises(
+            share_exceptions.AccessRuleBuildErrorException,
+            self.alt_shares_v2_client.wait_for_access_rule_status,
+            share2['id'], access2['id'], 'active')
 
 
 @ddt.ddt
@@ -427,7 +438,6 @@
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
     @ddt.data('shares_client', 'shares_v2_client')
     def test_delete_access_rule_with_wrong_id(self, client_name):
-        skip_if_cephx_access_type_not_supported_by_client(self, client_name)
         self.assertRaises(lib_exc.NotFound,
                           getattr(self, client_name).delete_access_rule,
                           self.share["id"], "wrong_rule_id")
@@ -435,7 +445,6 @@
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
     @ddt.data('shares_client', 'shares_v2_client')
     def test_create_access_rule_ip_with_wrong_type(self, client_name):
-        skip_if_cephx_access_type_not_supported_by_client(self, client_name)
         self.assertRaises(lib_exc.BadRequest,
                           getattr(self, client_name).create_access_rule,
                           self.share["id"], "wrong_type", "1.2.3.4")
@@ -445,7 +454,6 @@
     @testtools.skipUnless(CONF.share.run_snapshot_tests,
                           "Snapshot tests are disabled.")
     def test_create_access_rule_ip_to_snapshot(self, client_name):
-        skip_if_cephx_access_type_not_supported_by_client(self, client_name)
         self.assertRaises(lib_exc.NotFound,
                           getattr(self, client_name).create_access_rule,
                           self.snap["id"])
@@ -457,7 +465,6 @@
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API)
     @ddt.data('shares_client', 'shares_v2_client')
     def test_create_access_rule_ip_with_wrong_share_id(self, client_name):
-        skip_if_cephx_access_type_not_supported_by_client(self, client_name)
         self.assertRaises(lib_exc.NotFound,
                           getattr(self, client_name).create_access_rule,
                           "wrong_share_id")
diff --git a/manila_tempest_tests/tests/api/test_shares_actions.py b/manila_tempest_tests/tests/api/test_shares_actions.py
index 7cedb6c..4829dfd 100644
--- a/manila_tempest_tests/tests/api/test_shares_actions.py
+++ b/manila_tempest_tests/tests/api/test_shares_actions.py
@@ -420,9 +420,7 @@
         )
 
         params = {"is_public": True}
-        isolated_client = self.get_client_with_isolated_creds(
-            type_of_creds='alt')
-        shares = isolated_client.list_shares_with_detail(params)
+        shares = self.alt_shares_client.list_shares_with_detail(params)
 
         keys = [
             "status", "description", "links", "availability_zone",
@@ -545,10 +543,10 @@
     @testtools.skipUnless(CONF.share.run_snapshot_tests,
                           "Snapshot tests are disabled.")
     def test_list_snapshots_with_detail_use_limit(self):
-        for l, o in [('1', '1'), ('0', '1')]:
+        for limit, offset in [('1', '1'), ('0', '1')]:
             filters = {
-                'limit': l,
-                'offset': o,
+                'limit': limit,
+                'offset': offset,
                 'share_id': self.shares[0]['id'],
             }
 
diff --git a/manila_tempest_tests/tests/api/test_shares_from_snapshot_across_pools.py b/manila_tempest_tests/tests/api/test_shares_from_snapshot_across_pools.py
index ce7f46d..d71202d 100644
--- a/manila_tempest_tests/tests/api/test_shares_from_snapshot_across_pools.py
+++ b/manila_tempest_tests/tests/api/test_shares_from_snapshot_across_pools.py
@@ -33,7 +33,7 @@
         # create share_type
         extra_specs = {"create_share_from_snapshot_support": True,
                        "snapshot_support": True}
-        cls.share_type = cls._create_share_type(extra_specs)
+        cls.share_type = cls._create_share_type(specs=extra_specs)
         cls.share_type_id = cls.share_type['id']
         cls.admin_client = cls.admin_shares_v2_client
         cls.pools = cls.get_pools_matching_share_type(cls.share_type,
@@ -135,8 +135,8 @@
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_BACKEND)
     def test_share_from_snapshot_across_azs(self):
-        azs = next(self.rep_domain_azs[rep] for rep in self.rep_domain_azs if
-                   len(self.rep_domain_azs[rep]) > 1)
+        azs = next((self.rep_domain_azs[rep] for rep in self.rep_domain_azs if
+                    len(self.rep_domain_azs[rep]) > 1), None)
         if azs is None:
             msg = ("Could not find the necessary azs. At least two azs "
                    "are needed to run the test to create share from snapshot "
diff --git a/manila_tempest_tests/tests/api/test_shares_negative.py b/manila_tempest_tests/tests/api/test_shares_negative.py
index a0a1a47..09a45cd 100644
--- a/manila_tempest_tests/tests/api/test_shares_negative.py
+++ b/manila_tempest_tests/tests/api/test_shares_negative.py
@@ -150,43 +150,35 @@
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
     def test_update_other_tenants_public_share(self):
-        isolated_client = self.get_client_with_isolated_creds(
-            type_of_creds='alt')
-        self.assertRaises(lib_exc.Forbidden, isolated_client.update_share,
-                          self.share["id"], name="new_name")
+        self.assertRaises(lib_exc.Forbidden,
+                          self.alt_shares_v2_client.update_share,
+                          self.share["id"],
+                          name="new_name")
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
     def test_delete_other_tenants_public_share(self):
-        isolated_client = self.get_client_with_isolated_creds(
-            type_of_creds='alt')
         self.assertRaises(lib_exc.Forbidden,
-                          isolated_client.delete_share,
+                          self.alt_shares_v2_client.delete_share,
                           self.share['id'])
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
     def test_set_metadata_of_other_tenants_public_share(self):
-        isolated_client = self.get_client_with_isolated_creds(
-            type_of_creds='alt')
         self.assertRaises(lib_exc.Forbidden,
-                          isolated_client.set_metadata,
+                          self.alt_shares_v2_client.set_metadata,
                           self.share['id'],
                           {'key': 'value'})
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
     def test_update_metadata_of_other_tenants_public_share(self):
-        isolated_client = self.get_client_with_isolated_creds(
-            type_of_creds='alt')
         self.assertRaises(lib_exc.Forbidden,
-                          isolated_client.update_all_metadata,
+                          self.alt_shares_v2_client.update_all_metadata,
                           self.share['id'],
                           {'key': 'value'})
 
     @tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
     def test_delete_metadata_of_other_tenants_public_share(self):
-        isolated_client = self.get_client_with_isolated_creds(
-            type_of_creds='alt')
         self.assertRaises(lib_exc.Forbidden,
-                          isolated_client.delete_metadata,
+                          self.alt_shares_v2_client.delete_metadata,
                           self.share['id'],
                           'key')
 
diff --git a/manila_tempest_tests/tests/api/test_snapshot_rules.py b/manila_tempest_tests/tests/api/test_snapshot_rules.py
index 86cb339..0323916 100644
--- a/manila_tempest_tests/tests/api/test_snapshot_rules.py
+++ b/manila_tempest_tests/tests/api/test_snapshot_rules.py
@@ -34,7 +34,7 @@
         super(BaseShareSnapshotRulesTest, cls).resource_setup()
         # create share_type
         extra_specs = {'mount_snapshot_support': 'True'}
-        cls.share_type = cls._create_share_type(extra_specs)
+        cls.share_type = cls._create_share_type(specs=extra_specs)
         cls.share_type_id = cls.share_type['id']
 
         # create share
diff --git a/manila_tempest_tests/tests/api/test_snapshot_rules_negative.py b/manila_tempest_tests/tests/api/test_snapshot_rules_negative.py
index 7a4308d..f9670db 100644
--- a/manila_tempest_tests/tests/api/test_snapshot_rules_negative.py
+++ b/manila_tempest_tests/tests/api/test_snapshot_rules_negative.py
@@ -49,7 +49,7 @@
         super(SnapshotIpRulesForNFSNegativeTest, cls).resource_setup()
         # create share type
         extra_specs = {'mount_snapshot_support': 'True'}
-        cls.share_type = cls._create_share_type(extra_specs)
+        cls.share_type = cls._create_share_type(specs=extra_specs)
         cls.share_type_id = cls.share_type['id']
         # create share
         cls.share = cls.create_share(cls.protocol,
diff --git a/manila_tempest_tests/tests/scenario/manager_share.py b/manila_tempest_tests/tests/scenario/manager_share.py
index da8a8e5..10074b1 100644
--- a/manila_tempest_tests/tests/scenario/manager_share.py
+++ b/manila_tempest_tests/tests/scenario/manager_share.py
@@ -124,6 +124,9 @@
     def mount_share(self, location, remote_client, target_dir=None):
         raise NotImplementedError
 
+    def allow_access(self, **kwargs):
+        raise NotImplementedError
+
     def unmount_share(self, remote_client, target_dir=None):
         target_dir = target_dir or "/mnt"
         remote_client.exec_command("sudo umount %s" % target_dir)
@@ -291,7 +294,7 @@
         if instance and not ip:
             try:
                 net_addresses = instance['addresses']
-                first_address = net_addresses.values()[0][0]
+                first_address = list(net_addresses.values())[0][0]
                 ip = first_address['addr']
             except Exception:
                 LOG.debug("Instance has no valid IP address: %s", instance)
@@ -347,6 +350,54 @@
                 share['id'], instance=instance, cleanup=False,
                 snapshot=snapshot, access_level=access_level, client=client)
 
+    def provide_access_to_client_identified_by_cephx(self, share=None,
+                                                     access_level='rw',
+                                                     access_to=None,
+                                                     remote_client=None,
+                                                     locations=None,
+                                                     client=None,
+                                                     oc_size=20971520):
+        share = share or self.share
+        client = client or self.shares_v2_client
+        access_to = access_to or data_utils.rand_name(
+            self.__class__.__name__ + '-cephx-id')
+        # Check if access is already granted to the client
+        access = self.shares_v2_client.list_access_rules(
+            share['id'], metadata={'metadata': {'access_to': access_to}})
+        access = access[0] if access else None
+
+        if not access:
+            access = self._allow_access(
+                share['id'], access_level=access_level, access_to=access_to,
+                access_type="cephx", cleanup=False, client=client)
+            # Set metadata to access rule to be filtered if necessary.
+            # This is necessary to prevent granting access to a client who
+            # already has.
+            self.shares_v2_client.update_access_metadata(
+                metadata={"access_to": "{}".format(access_to)},
+                access_id=access['id'])
+        get_access = self.shares_v2_client.get_access(access['id'])
+        # Set 'access_key' and 'access_to' attributes for being use in mount
+        # operation.
+        setattr(self, 'access_key', get_access['access_key'])
+        setattr(self, 'access_to', access_to)
+
+        remote_client.exec_command(
+            "sudo crudini --set {access_to}.keyring client.{access_to} key "
+            "{access_key}"
+            .format(access_to=access_to, access_key=self.access_key))
+        remote_client.exec_command(
+            "sudo crudini --set ceph.conf client \"client quota\" true")
+        remote_client.exec_command(
+            "sudo crudini --set ceph.conf client \"client oc size\" {}"
+            .format(oc_size))
+        if not isinstance(locations, list):
+            locations = [locations]
+        remote_client.exec_command(
+            "sudo crudini --set ceph.conf client \"mon host\" {}"
+            .format(locations[0].split(':/')[0]))
+        return access
+
     def wait_for_active_instance(self, instance_id):
         waiters.wait_for_server_status(
             self.os_primary.servers_client, instance_id, "ACTIVE")
@@ -598,9 +649,10 @@
             locations = self._get_snapshot_export_locations(snapshot)
 
         self.assertNotEmpty(locations)
-        locations = self._get_export_locations_according_to_ip_version(
-            locations, error_on_invalid_ip_version)
-        self.assertNotEmpty(locations)
+        if self.protocol != 'cephfs':
+            locations = self._get_export_locations_according_to_ip_version(
+                locations, error_on_invalid_ip_version)
+            self.assertNotEmpty(locations)
 
         return locations
 
@@ -630,3 +682,39 @@
             message = ("Protocol %s is not supported" % self.protocol)
             raise self.skipException(message)
         return ip, version
+
+
+class BaseShareCEPHFSTest(ShareScenarioTest):
+
+    def allow_access(self, access_level='rw', **kwargs):
+        return self.provide_access_to_client_identified_by_cephx(
+            remote_client=kwargs['remote_client'],
+            locations=kwargs['locations'], access_level=access_level)
+
+    def _fuse_client(self, mountpoint, remote_client, target_dir, access_to):
+        remote_client.exec_command(
+            "sudo ceph-fuse {target_dir} --id={access_to} --conf=ceph.conf "
+            "--keyring={access_to}.keyring --client-mountpoint={mountpoint}"
+            .format(target_dir=target_dir, access_to=access_to,
+                    mountpoint=mountpoint))
+
+    def mount_share(self, location, remote_client, target_dir=None,
+                    access_to=None):
+        target_dir = target_dir or "/mnt"
+        access_to = access_to or self.access_to
+        mountpoint = location.split(':')[-1]
+        if getattr(self, 'mount_client', None):
+            return self._fuse_client(mountpoint, remote_client, target_dir,
+                                     access_to=access_to)
+        remote_client.exec_command(
+            "sudo mount -t ceph {location} {target_dir} -o name={access_to},"
+            "secret={access_key}"
+            .format(location=location, target_dir=target_dir,
+                    access_to=access_to, access_key=self.access_key))
+
+    def unmount_share(self, remote_client, target_dir=None):
+        target_dir = target_dir or "/mnt"
+        if getattr(self, 'mount_client', None):
+            return remote_client.exec_command(
+                "sudo fusermount -uz %s" % target_dir)
+        super(BaseShareCEPHFSTest, self).unmount_share(remote_client)
diff --git a/manila_tempest_tests/tests/scenario/test_share_basic_ops.py b/manila_tempest_tests/tests/scenario/test_share_basic_ops.py
index 8a55b2e..ef73919 100644
--- a/manila_tempest_tests/tests/scenario/test_share_basic_ops.py
+++ b/manila_tempest_tests/tests/scenario/test_share_basic_ops.py
@@ -44,14 +44,6 @@
      * Terminate the instance
     """
 
-    @classmethod
-    def skip_checks(cls):
-        super(ShareBasicOpsBase, cls).skip_checks()
-        if cls.protocol not in CONF.share.enable_ip_rules_for_protocols:
-            message = ("%s tests for access rules other than IP are disabled" %
-                       cls.protocol)
-            raise cls.skipException(message)
-
     def _ping_host_from_export_location(self, export, remote_client):
         ip, version = self.get_ip_and_version_from_export_location(export)
         if version == 6:
@@ -66,7 +58,8 @@
         locations = self.get_user_export_locations(self.share)
         instance = self.wait_for_active_instance(instance["id"])
         remote_client = self.init_remote_client(instance)
-        self.provide_access_to_auxiliary_instance(instance)
+        self.allow_access(instance=instance, remote_client=remote_client,
+                          locations=locations)
 
         for location in locations:
             self.mount_share(location, remote_client)
@@ -85,12 +78,17 @@
         remote_client_inst = self.init_remote_client(instance)
 
         # First, check if write works RW access.
-        acc_rule_id = self.provide_access_to_auxiliary_instance(instance)['id']
+        acc_rule_id = self.allow_access(
+            instance=instance, remote_client=remote_client_inst,
+            locations=location)['id']
+
         self.mount_share(location, remote_client_inst)
         self.write_data_to_mounted_share(test_data, remote_client_inst)
         self.deny_access(self.share['id'], acc_rule_id)
 
-        self.provide_access_to_auxiliary_instance(instance, access_level='ro')
+        self.allow_access(instance=instance, remote_client=remote_client_inst,
+                          locations=location, access_level='ro')
+
         self.addCleanup(self.unmount_share, remote_client_inst)
 
         # Test if write with RO access fails.
@@ -113,7 +111,9 @@
 
         # Write data to first VM
         remote_client_inst1 = self.init_remote_client(instance1)
-        self.provide_access_to_auxiliary_instance(instance1)
+        self.allow_access(instance=instance1,
+                          remote_client=remote_client_inst1,
+                          locations=location)
 
         self.mount_share(location, remote_client_inst1)
         self.addCleanup(self.unmount_share,
@@ -123,7 +123,9 @@
         # Read from second VM
         remote_client_inst2 = self.init_remote_client(instance2)
         if not CONF.share.override_ip_for_nfs_access or self.ipv6_enabled:
-            self.provide_access_to_auxiliary_instance(instance2)
+            self.allow_access(instance=instance2,
+                              remote_client=remote_client_inst2,
+                              locations=location)
 
         self.mount_share(location, remote_client_inst2)
         self.addCleanup(self.unmount_share,
@@ -241,7 +243,8 @@
 
     @tc.attr(base.TAG_POSITIVE, base.TAG_BACKEND)
     @testtools.skipUnless(
-        CONF.share.run_snapshot_tests, "Snapshot tests are disabled.")
+        CONF.share.capability_create_share_from_snapshot_support,
+        "Create share from snapshot tests are disabled.")
     def test_write_data_to_share_created_from_snapshot(self):
         # 1 - Create UVM, ok, created
         instance = self.boot_instance(wait_until="BUILD")
@@ -387,6 +390,18 @@
 class TestShareBasicOpsNFS(ShareBasicOpsBase):
     protocol = "nfs"
 
+    @classmethod
+    def skip_checks(cls):
+        super(TestShareBasicOpsNFS, cls).skip_checks()
+        if cls.protocol not in CONF.share.enable_ip_rules_for_protocols:
+            message = ("%s tests for access rules other than IP are disabled" %
+                       cls.protocol)
+            raise cls.skipException(message)
+
+    def allow_access(self, access_level='rw', **kwargs):
+        return self.provide_access_to_auxiliary_instance(
+            instance=kwargs['instance'], access_level=access_level)
+
     def mount_share(self, location, remote_client, target_dir=None):
 
         self._ping_host_from_export_location(location, remote_client)
@@ -399,6 +414,18 @@
 class TestShareBasicOpsCIFS(ShareBasicOpsBase):
     protocol = "cifs"
 
+    @classmethod
+    def skip_checks(cls):
+        super(TestShareBasicOpsCIFS, cls).skip_checks()
+        if cls.protocol not in CONF.share.enable_ip_rules_for_protocols:
+            message = ("%s tests for access rules other than IP are disabled" %
+                       cls.protocol)
+            raise cls.skipException(message)
+
+    def allow_access(self, access_level='rw', **kwargs):
+        return self.provide_access_to_auxiliary_instance(
+            instance=kwargs['instance'], access_level=access_level)
+
     def mount_share(self, location, remote_client, target_dir=None):
 
         self._ping_host_from_export_location(location, remote_client)
@@ -426,6 +453,25 @@
         raise self.skipException(msg)
 
 
+class TestShareBasicOpsCEPHFS(ShareBasicOpsBase, manager.BaseShareCEPHFSTest):
+    protocol = "cephfs"
+
+    @tc.attr(base.TAG_POSITIVE, base.TAG_BACKEND)
+    def test_mount_share_one_vm_with_ceph_fuse_client(self):
+        self.mount_client = 'fuse'
+        super(TestShareBasicOpsCEPHFS, self).test_mount_share_one_vm()
+
+    @tc.attr(base.TAG_POSITIVE, base.TAG_BACKEND)
+    def test_write_with_ro_access_with_ceph_fuse_client(self):
+        self.mount_client = 'fuse'
+        super(TestShareBasicOpsCEPHFS, self).test_write_with_ro_access()
+
+    @tc.attr(base.TAG_POSITIVE, base.TAG_BACKEND)
+    def test_read_write_two_vms_with_ceph_fuse_client(self):
+        self.mount_client = 'fuse'
+        super(TestShareBasicOpsCEPHFS, self).test_read_write_two_vms()
+
+
 class TestShareBasicOpsNFSIPv6(TestShareBasicOpsNFS):
     ip_version = 6
 
diff --git a/manila_tempest_tests/tests/scenario/test_share_extend.py b/manila_tempest_tests/tests/scenario/test_share_extend.py
index 7827b75..e0a1e4a 100644
--- a/manila_tempest_tests/tests/scenario/test_share_extend.py
+++ b/manila_tempest_tests/tests/scenario/test_share_extend.py
@@ -45,14 +45,6 @@
      * Terminate the instance
     """
 
-    @classmethod
-    def skip_checks(cls):
-        super(ShareExtendBase, cls).skip_checks()
-        if cls.protocol not in CONF.share.enable_ip_rules_for_protocols:
-            message = ("%s tests for access rules other than IP are disabled" %
-                       cls.protocol)
-            raise cls.skipException(message)
-
     @tc.attr(base.TAG_POSITIVE, base.TAG_BACKEND)
     def test_create_extend_and_write(self):
         default_share_size = CONF.share.share_size
@@ -69,10 +61,11 @@
         remote_client = self.init_remote_client(instance)
 
         LOG.debug('Step 4 - grant access')
-        self.provide_access_to_auxiliary_instance(instance, share=share)
+        location = self.get_share_export_location_for_mount(share)
+        self.allow_access(instance=instance, remote_client=remote_client,
+                          locations=location)
 
         LOG.debug('Step 5 - mount')
-        location = self.get_share_export_location_for_mount(share)
         self.mount_share(location, remote_client)
 
         total_blocks = (units.Ki * default_share_size) / 64
@@ -153,6 +146,18 @@
 class TestShareExtendNFS(ShareExtendBase):
     protocol = "nfs"
 
+    @classmethod
+    def skip_checks(cls):
+        super(ShareExtendBase, cls).skip_checks()
+        if cls.protocol not in CONF.share.enable_ip_rules_for_protocols:
+            message = ("%s tests for access rules other than IP are disabled" %
+                       cls.protocol)
+            raise cls.skipException(message)
+
+    def allow_access(self, access_level='rw', **kwargs):
+        return self.provide_access_to_auxiliary_instance(
+            instance=kwargs['instance'], access_level=access_level)
+
     def mount_share(self, location, remote_client, target_dir=None):
         target_dir = target_dir or "/mnt"
         remote_client.exec_command(
@@ -163,6 +168,18 @@
 class TestShareExtendCIFS(ShareExtendBase):
     protocol = "cifs"
 
+    @classmethod
+    def skip_checks(cls):
+        super(ShareExtendBase, cls).skip_checks()
+        if cls.protocol not in CONF.share.enable_ip_rules_for_protocols:
+            message = ("%s tests for access rules other than IP are disabled" %
+                       cls.protocol)
+            raise cls.skipException(message)
+
+    def allow_access(self, access_level='rw', **kwargs):
+        return self.provide_access_to_auxiliary_instance(
+            instance=kwargs['instance'], access_level=access_level)
+
     def mount_share(self, location, remote_client, target_dir=None):
         location = location.replace("\\", "/")
         target_dir = target_dir or "/mnt"
@@ -171,6 +188,15 @@
         )
 
 
+class TestShareExtendCEPHFS(ShareExtendBase, manager.BaseShareCEPHFSTest):
+    protocol = "cephfs"
+
+    @tc.attr(base.TAG_POSITIVE, base.TAG_BACKEND)
+    def test_create_extend_and_write_with_ceph_fuse_client(self):
+        self.mount_client = 'fuse'
+        super(TestShareExtendCEPHFS, self).test_create_extend_and_write()
+
+
 # NOTE(u_glide): this function is required to exclude ShareExtendBase
 # from executed test cases.
 # See: https://docs.python.org/3/library/unittest.html#load-tests-protocol
diff --git a/manila_tempest_tests/tests/scenario/test_share_shrink.py b/manila_tempest_tests/tests/scenario/test_share_shrink.py
index e8ffdcb..785c6ce 100644
--- a/manila_tempest_tests/tests/scenario/test_share_shrink.py
+++ b/manila_tempest_tests/tests/scenario/test_share_shrink.py
@@ -46,14 +46,6 @@
      * Terminate the instance
     """
 
-    @classmethod
-    def skip_checks(cls):
-        super(ShareShrinkBase, cls).skip_checks()
-        if cls.protocol not in CONF.share.enable_ip_rules_for_protocols:
-            message = ("%s tests for access rules other than IP are disabled" %
-                       cls.protocol)
-            raise cls.skipException(message)
-
     @tc.attr(base.TAG_POSITIVE, base.TAG_BACKEND)
     @testtools.skipUnless(
         CONF.share.run_shrink_tests, 'Shrink share tests are disabled.')
@@ -72,10 +64,11 @@
         remote_client = self.init_remote_client(instance)
 
         LOG.debug('Step 4 - grant access')
-        self.provide_access_to_auxiliary_instance(instance)
+        location = self.get_share_export_location_for_mount(share)
+        self.allow_access(instance=instance, remote_client=remote_client,
+                          locations=location)
 
         LOG.debug('Step 5 - mount')
-        location = self.get_share_export_location_for_mount(share)
         self.mount_share(location, remote_client)
 
         total_blocks = (1024 * default_share_size) / 64
@@ -167,6 +160,18 @@
 class TestShareShrinkNFS(ShareShrinkBase):
     protocol = "nfs"
 
+    @classmethod
+    def skip_checks(cls):
+        super(ShareShrinkBase, cls).skip_checks()
+        if cls.protocol not in CONF.share.enable_ip_rules_for_protocols:
+            message = ("%s tests for access rules other than IP are disabled" %
+                       cls.protocol)
+            raise cls.skipException(message)
+
+    def allow_access(self, access_level='rw', **kwargs):
+        return self.provide_access_to_auxiliary_instance(
+            instance=kwargs['instance'], access_level=access_level)
+
     def mount_share(self, location, ssh_client, target_dir=None):
         target_dir = target_dir or "/mnt"
         ssh_client.exec_command(
@@ -177,6 +182,18 @@
 class TestShareShrinkCIFS(ShareShrinkBase):
     protocol = "cifs"
 
+    @classmethod
+    def skip_checks(cls):
+        super(ShareShrinkBase, cls).skip_checks()
+        if cls.protocol not in CONF.share.enable_ip_rules_for_protocols:
+            message = ("%s tests for access rules other than IP are disabled" %
+                       cls.protocol)
+            raise cls.skipException(message)
+
+    def allow_access(self, access_level='rw', **kwargs):
+        return self.provide_access_to_auxiliary_instance(
+            instance=kwargs['instance'], access_level=access_level)
+
     def mount_share(self, location, ssh_client, target_dir=None):
         location = location.replace("\\", "/")
         target_dir = target_dir or "/mnt"
@@ -185,6 +202,15 @@
         )
 
 
+class TestShareShrinkCEPHFS(ShareShrinkBase, manager.BaseShareCEPHFSTest):
+    protocol = "cephfs"
+
+    @tc.attr(base.TAG_POSITIVE, base.TAG_BACKEND)
+    def test_create_shrink_and_write_with_ceph_fuse_client(self):
+        self.mount_client = 'fuse'
+        super(TestShareShrinkCEPHFS, self).test_create_shrink_and_write()
+
+
 # NOTE(u_glide): this function is required to exclude ShareShrinkBase from
 # executed test cases.
 # See: https://docs.python.org/3/library/unittest.html#load-tests-protocol
diff --git a/test-requirements.txt b/test-requirements.txt
index e3eead6..f230fa9 100644
--- a/test-requirements.txt
+++ b/test-requirements.txt
@@ -2,7 +2,7 @@
 # of appearance. Changing the order has an impact on the overall integration
 # process, which may cause wedges in the gate later.
 
-hacking>=3.0,<3.1.0;python_version>'3' # Apache-2.0
+hacking>=3.0.1,<3.1.0;python_version>'3' # Apache-2.0
 
 coverage!=4.4,>=4.0 # Apache-2.0
 python-subunit>=1.0.0 # Apache-2.0/BSD
diff --git a/zuul.d/manila-tempest-jobs.yaml b/zuul.d/manila-tempest-jobs.yaml
index c9efb7c..95a5f60 100644
--- a/zuul.d/manila-tempest-jobs.yaml
+++ b/zuul.d/manila-tempest-jobs.yaml
@@ -4,13 +4,12 @@
     description: Base job for devstack/tempest based manila jobs.
     parent: devstack-tempest
     timeout: 10800
-    required-projects:
+    required-projects: &manila-tempest-required-projects
       - openstack/manila
-      - openstack/manila-image-elements
       - openstack/manila-tempest-plugin
       - openstack/python-manilaclient
       - openstack/tempest
-    irrelevant-files:
+    irrelevant-files: &irrelevant-files
       - ^(test-|)requirements.txt$
       - ^.*\.rst$
       - ^api-ref/.*$
@@ -21,7 +20,7 @@
       - ^setup.cfg$
       - ^tools/.*$
       - ^tox.ini$
-    vars:
+    vars: &manila-tempest-base-vars
       tox_envlist: all
       tempest_test_regex: manila_tempest_tests
       tempest_concurrency: 8
@@ -40,14 +39,26 @@
         tls-proxy: true
       devstack_localrc:
         USE_PYTHON3: true
-        TEMPEST_USE_TEST_ACCOUNTS: true
         MANILA_USE_DOWNGRADE_MIGRATIONS: true
         MANILA_INSTALL_TEMPEST_PLUGIN_SYSTEMWIDE: false
         MANILA_ALLOW_NAS_SERVER_PORTS_ON_HOST: true
         MANILA_DEFAULT_SHARE_TYPE_EXTRA_SPECS: 'snapshot_support=True create_share_from_snapshot_support=True'
+        MANILA_CONFIGURE_DEFAULT_TYPES: true
 
 
 - job:
+    name: manila-tempest-plugin-ipv6-base
+    abstract: true
+    description: |
+      Base job for devstack/tempest based manila jobs in a 4+6 (dual-stack)
+      devstack environment with IPv6 control plane endpoints.
+    parent: devstack-tempest-ipv6
+    timeout: 10800
+    irrelevant-files: *irrelevant-files
+    required-projects: *manila-tempest-required-projects
+    vars: *manila-tempest-base-vars
+
+- job:
     name: manila-tempest-plugin-zfsonlinux
     description: Test ZFSOnLinux multibackend (DHSS=False) with postgresql db
     parent: manila-tempest-plugin-base
@@ -66,7 +77,6 @@
         MANILA_REPLICA_STATE_UPDATE_INTERVAL: 60
         MANILA_ZFSONLINUX_SERVICE_IP: 127.0.0.1
         MANILA_ZFSONLINUX_USE_SSH: true
-        MANILA_CONFIGURE_DEFAULT_TYPES: true
         MANILA_USE_SCHEDULER_CREATING_SHARE_FROM_SNAPSHOT: true
       devstack_services:
         mysql: false
@@ -92,3 +102,168 @@
               backend_names: LONDON,PARIS
               multi_backend: true
               image_password: manila
+
+- job:
+    name: manila-tempest-plugin-lvm
+    description: |
+      Test LVM multibackend (DHSS=False) in a 4+6 (dual-stack) devstack
+      environment with IPv6 control plane endpoints.
+    parent: manila-tempest-plugin-ipv6-base
+    required-projects:
+      - openstack/neutron-dynamic-routing
+    vars:
+      tempest_test_regex: '(^manila_tempest_tests.tests.api)(?=.*\[.*\bbackend\b.*\])'
+      devstack_localrc:
+        SHARE_DRIVER: manila.share.drivers.lvm.LVMShareDriver
+        MANILA_ENABLED_BACKENDS: london,paris
+        MANILA_BACKEND1_CONFIG_GROUP_NAME: london
+        MANILA_BACKEND2_CONFIG_GROUP_NAME: paris
+        MANILA_SHARE_BACKEND1_NAME: LONDON
+        MANILA_SHARE_BACKEND2_NAME: PARIS
+        MANILA_OPTGROUP_london_driver_handles_share_servers: false
+        MANILA_OPTGROUP_paris_driver_handles_share_servers: false
+        SHARE_BACKING_FILE_SIZE: 32000M
+        MANILA_SHARE_MIGRATION_PERIOD_TASK_INTERVAL: 1
+        MANILA_DEFAULT_SHARE_TYPE_EXTRA_SPECS: 'snapshot_support=True create_share_from_snapshot_support=True revert_to_snapshot_support=True mount_snapshot_support=True'
+        MANILA_SETUP_IPV6: true
+        NEUTRON_CREATE_INITIAL_NETWORKS: false
+        MANILA_RESTORE_IPV6_DEFAULT_ROUTE: false
+      devstack_plugins:
+        neutron-dynamic-routing: https://opendev.org/openstack/neutron-dynamic-routing
+      devstack_local_conf:
+        test-config:
+          $TEMPEST_CONFIG:
+            share:
+              default_share_type_name: default
+              run_ipv6_tests: true
+              run_mount_snapshot_tests: true
+              run_host_assisted_migration_tests: true
+              run_shrink_tests: false
+              run_revert_to_snapshot_tests: true
+              enable_user_rules_for_protocols: cifs
+              enable_ip_rules_for_protocols: nfs
+              multitenancy_enabled: False
+              backend_names: LONDON,PARIS
+              multi_backend: true
+              image_password: manila
+
+- job:
+    name: manila-tempest-plugin-container
+    description: |
+      Test the container driver multibackend (DHSS=True) with CIFS
+      and postgresql db. API only, in a minimal devstack
+    parent: manila-tempest-plugin-base
+    vars:
+      tempest_test_regex: manila_tempest_tests.tests.api
+      devstack_localrc:
+        SHARE_DRIVER: manila.share.drivers.container.driver.ContainerShareDriver
+        MANILA_SERVICE_IMAGE_ENABLED: false
+        MANILA_ENABLED_BACKENDS: london,paris
+        MANILA_BACKEND1_CONFIG_GROUP_NAME: london
+        MANILA_BACKEND2_CONFIG_GROUP_NAME: paris
+        MANILA_SHARE_BACKEND1_NAME: LONDON
+        MANILA_SHARE_BACKEND2_NAME: PARIS
+        MANILA_OPTGROUP_london_driver_handles_share_servers: true
+        MANILA_OPTGROUP_paris_driver_handles_share_servers: true
+        SHARE_BACKING_FILE_SIZE: 64000M
+        MANILA_CONFIGURE_DEFAULT_TYPES: false
+        MANILA_DEFAULT_SHARE_TYPE_EXTRA_SPECS: 'snapshot_support=false'
+        TEMPEST_USE_TEST_ACCOUNTS: true
+      devstack_services:
+        mysql: false
+        postgresql: true
+      devstack_local_conf:
+        test-config:
+          $TEMPEST_CONFIG:
+            share:
+              enable_ip_rules_for_protocols: ""
+              run_share_group_tests: false
+              run_manage_unmanage_tests: true
+              run_snapshot_tests: false
+              run_shrink_tests: false
+              enable_user_rules_for_protocols: cifs
+              enable_protocols: cifs
+              capability_storage_protocol: CIFS
+              multitenancy_enabled: true
+              backend_names: LONDON,PARIS
+              multi_backend: true
+
+- job:
+    name: manila-tempest-plugin-generic
+    description: |
+      Test the generic driver multibackend (DHSS=True) with NFS and CIFS
+    parent: manila-tempest-plugin-base
+    vars:
+      tempest_test_regex: '(^manila_tempest_tests.tests)(?=.*\[.*\bbackend\b.*\])'
+      # The generic driver uses nova VMs as share servers; running with a
+      # high concurrency could starve the driver of RAM/Disk/CPUs to
+      # function properly in a small single node devstack VM.
+      tempest_concurrency: 2
+      devstack_localrc:
+        VOLUME_BACKING_FILE_SIZE: 24G
+        SHARE_DRIVER: manila.share.drivers.generic.GenericShareDriver
+        MANILA_ENABLED_BACKENDS: london,paris
+        MANILA_BACKEND1_CONFIG_GROUP_NAME: london
+        MANILA_BACKEND2_CONFIG_GROUP_NAME: paris
+        MANILA_SHARE_BACKEND1_NAME: LONDON
+        MANILA_SHARE_BACKEND2_NAME: PARIS
+        MANILA_OPTGROUP_london_driver_handles_share_servers: true
+        MANILA_OPTGROUP_paris_driver_handles_share_servers: true
+        MANILA_USE_SERVICE_INSTANCE_PASSWORD: true
+        MANILA_DEFAULT_SHARE_TYPE_EXTRA_SPECS: 'snapshot_support=True create_share_from_snapshot_support=True'
+        TEMPEST_USE_TEST_ACCOUNTS: true
+      devstack_services:
+        cinder: true
+      devstack_local_conf:
+        test-config:
+          $TEMPEST_CONFIG:
+            share:
+              default_share_type_name: default
+              run_share_group_tests: false
+              multitenancy_enabled: true
+              suppress_errors_in_cleanup: true
+              share_creation_retry_number: 2
+              backend_names: LONDON,PARIS
+              multi_backend: true
+              image_password: manila
+
+- job:
+    name: manila-tempest-plugin-cephfs-native
+    description: Test CephFS Native (DHSS=False)
+    parent: manila-tempest-plugin-base
+    required-projects:
+      - openstack/devstack-plugin-ceph
+    vars:
+      devstack_plugins:
+        devstack-plugin-ceph: https://opendev.org/openstack/devstack-plugin-ceph
+      tempest_test_regex: manila_tempest_tests
+      # Those tests fail when using kernel client, because right now CephFS quotas
+      # is not supported. So for now those tests should be blacklisted.
+      # TODO(lkuchlan) un-blacklist those test when https://review.opendev.org/#/c/676722 will land.
+      tempest_black_regex: '(TestShareShrinkCEPHFS.test_create_shrink_and_write|TestShareExtendCEPHFS.test_create_extend_and_write)(\[.*\bbackend\b.*\])'
+      devstack_localrc:
+        SHARE_DRIVER: manila.share.drivers.cephfs.driver.CephFSDriver
+        MANILA_ENABLED_BACKENDS: cephfsnative
+        ENABLED_SHARE_PROTOCOLS: CEPHFS
+        MANILA_DEFAULT_SHARE_TYPE_EXTRA_SPECS: 'snapshot_support=False'
+        MANILA_OPTGROUP_cephfsnative_share_driver: manila.share.drivers.cephfs.driver.CephFSDriver
+        MANILA_OPTGROUP_cephfsnative_driver_handles_share_servers: false
+        MANILA_OPTGROUP_cephfsnative_cephfs_conf_path: /etc/ceph/ceph.conf
+        MANILA_OPTGROUP_cephfsnative_cephfs_auth_id: manila
+        MANILA_SERVICE_IMAGE_URL: https://tarballs.opendev.org/openstack/manila-image-elements/images/manila-service-image-cephfs-master.qcow2
+        MANILA_SERVICE_IMAGE_NAME: manila-service-image-cephfs-master
+      devstack_local_conf:
+        test-config:
+          $TEMPEST_CONFIG:
+            share:
+              default_share_type_name: default
+              enable_cephx_rules_for_protocols: cephfs
+              run_share_group_tests: false
+              run_snapshot_tests: false
+              multitenancy_enabled: false
+              suppress_errors_in_cleanup: true
+              backend_names: CEPHFSNATIVE
+              capability_snapshot_support: false
+              capability_storage_protocol: CEPHFS
+              enable_protocols: cephfs
+              image_password: manila
diff --git a/zuul.d/project.yaml b/zuul.d/project.yaml
index bc0d8d3..e06e26d 100644
--- a/zuul.d/project.yaml
+++ b/zuul.d/project.yaml
@@ -4,29 +4,23 @@
       - tempest-plugin-jobs
     check:
       jobs:
+        - manila-tempest-plugin-lvm
+        - manila-tempest-minimal-dsvm-dummy
         - manila-tempest-plugin-zfsonlinux:
             voting: false
-        - manila-tempest-dsvm-mysql-generic:
+        - manila-tempest-plugin-cephfs-native:
             voting: false
-        - manila-tempest-dsvm-postgres-container:
+        - manila-tempest-plugin-container:
             voting: false
-        - manila-tempest-dsvm-postgres-generic-singlebackend:
-            voting: false
-        - manila-tempest-dsvm-generic-no-share-servers:
-            voting: false
-        - manila-tempest-dsvm-scenario:
-            voting: false
-        - manila-tempest-minimal-dsvm-cephfs-native:
+        - manila-tempest-plugin-generic:
             voting: false
         - manila-tempest-minimal-dsvm-cephfs-nfs:
             voting: false
-        - manila-tempest-minimal-dsvm-dummy
-        - manila-tempest-minimal-dsvm-lvm
     gate:
       queue: manila
       jobs:
         - manila-tempest-minimal-dsvm-dummy
-        - manila-tempest-minimal-dsvm-lvm
+        - manila-tempest-plugin-lvm
     experimental:
       jobs:
         - manila-tempest-dsvm-glusterfs-nfs: