diff --git a/releasenotes/notes/add-image-task-apis-as-tempest-clients-228ccba01f59cbf3.yaml b/releasenotes/notes/add-image-task-apis-as-tempest-clients-228ccba01f59cbf3.yaml
new file mode 100644
index 0000000..cb99a29
--- /dev/null
+++ b/releasenotes/notes/add-image-task-apis-as-tempest-clients-228ccba01f59cbf3.yaml
@@ -0,0 +1,54 @@
+---
+features:
+  - |
+    The following ``tasks_client`` tempest client for glance v2 image
+    task API is implemented in this release.
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/tempest/api/compute/admin/test_live_migration.py b/tempest/api/compute/admin/test_live_migration.py
index 19026d3..a02820a 100644
--- a/tempest/api/compute/admin/test_live_migration.py
+++ b/tempest/api/compute/admin/test_live_migration.py
@@ -304,13 +304,17 @@
     min_microversion = '2.6'
     max_microversion = 'latest'
 
+    @classmethod
+    def skip_checks(cls):
+        super(LiveMigrationRemoteConsolesV26Test, cls).skip_checks()
+        if not CONF.compute_feature_enabled.serial_console:
+            skip_msg = ("Serial console not supported.")
+            raise cls.skipException(skip_msg)
+        if not compute.is_scheduler_filter_enabled("DifferentHostFilter"):
+            raise cls.skipException("DifferentHostFilter is not available.")
+
     @decorators.attr(type='multinode')
     @decorators.idempotent_id('6190af80-513e-4f0f-90f2-9714e84955d7')
-    @testtools.skipUnless(CONF.compute_feature_enabled.serial_console,
-                          'Serial console not supported.')
-    @testtools.skipUnless(
-        compute.is_scheduler_filter_enabled("DifferentHostFilter"),
-        'DifferentHostFilter is not available.')
     def test_live_migration_serial_console(self):
         """Test the live-migration of an instance which has a serial console
 
diff --git a/tempest/api/compute/admin/test_volume_swap.py b/tempest/api/compute/admin/test_volume_swap.py
index 36148c5..9576b74 100644
--- a/tempest/api/compute/admin/test_volume_swap.py
+++ b/tempest/api/compute/admin/test_volume_swap.py
@@ -13,7 +13,6 @@
 import time
 
 from tempest.api.compute import base
-from tempest.common import utils
 from tempest.common import waiters
 from tempest import config
 from tempest.lib import decorators
@@ -33,6 +32,8 @@
     @classmethod
     def skip_checks(cls):
         super(TestVolumeSwapBase, cls).skip_checks()
+        if not CONF.service_available.cinder:
+            raise cls.skipException("Cinder is not available")
         if not CONF.compute_feature_enabled.swap_volume:
             raise cls.skipException("Swapping volumes is not supported.")
 
@@ -81,7 +82,6 @@
     # so it's marked as such.
     @decorators.attr(type='slow')
     @decorators.idempotent_id('1769f00d-a693-4d67-a631-6a3496773813')
-    @utils.services('volume')
     def test_volume_swap(self):
         """Test swapping of volume attached to server with admin user
 
@@ -183,7 +183,6 @@
     # multiple computes but that would just side-step the underlying bug.
     @decorators.skip_because(bug='1807723',
                              condition=CONF.compute.min_compute_nodes > 1)
-    @utils.services('volume')
     def test_volume_swap_with_multiattach(self):
         """Test swapping volume attached to multiple servers
 
diff --git a/tempest/api/compute/base.py b/tempest/api/compute/base.py
index ce6cd60..9aa5d35 100644
--- a/tempest/api/compute/base.py
+++ b/tempest/api/compute/base.py
@@ -51,6 +51,9 @@
         super(BaseV2ComputeTest, cls).skip_checks()
         if not CONF.service_available.nova:
             raise cls.skipException("Nova is not available")
+        if cls.create_default_network and not CONF.service_available.neutron:
+            raise cls.skipException("Neutron is not available")
+
         api_version_utils.check_skip_with_microversion(
             cls.min_microversion, cls.max_microversion,
             CONF.compute.min_microversion, CONF.compute.max_microversion)
diff --git a/tempest/api/compute/images/test_image_metadata_negative.py b/tempest/api/compute/images/test_image_metadata_negative.py
index b9806c7..33a59ae 100644
--- a/tempest/api/compute/images/test_image_metadata_negative.py
+++ b/tempest/api/compute/images/test_image_metadata_negative.py
@@ -14,10 +14,13 @@
 #    under the License.
 
 from tempest.api.compute import base
+from tempest import config
 from tempest.lib.common.utils import data_utils
 from tempest.lib import decorators
 from tempest.lib import exceptions as lib_exc
 
+CONF = config.CONF
+
 
 class ImagesMetadataNegativeTestJSON(base.BaseV2ComputeTest):
     """Negative tests of image metadata
@@ -28,6 +31,13 @@
     max_microversion = '2.38'
 
     @classmethod
+    def skip_checks(cls):
+        super(ImagesMetadataNegativeTestJSON, cls).skip_checks()
+        if not CONF.service_available.glance:
+            skip_msg = ("%s skipped as glance is not available" % cls.__name__)
+            raise cls.skipException(skip_msg)
+
+    @classmethod
     def setup_clients(cls):
         super(ImagesMetadataNegativeTestJSON, cls).setup_clients()
         cls.client = cls.compute_images_client
diff --git a/tempest/api/compute/servers/test_server_actions.py b/tempest/api/compute/servers/test_server_actions.py
index 59cf360..81f9e55 100644
--- a/tempest/api/compute/servers/test_server_actions.py
+++ b/tempest/api/compute/servers/test_server_actions.py
@@ -810,9 +810,15 @@
 
     min_microversion = '2.47'
 
+    @classmethod
+    def skip_checks(cls):
+        if not CONF.service_available.glance:
+            skip_msg = ("%s skipped as glance is not available" % cls.__name__)
+            raise cls.skipException(skip_msg)
+        super(ServersAaction247Test, cls).skip_checks()
+
     @testtools.skipUnless(CONF.compute_feature_enabled.snapshot,
                           'Snapshotting not available, backup not possible.')
-    @utils.services('image')
     @decorators.idempotent_id('252a4bdd-6366-4dae-9994-8c30aa660f23')
     def test_create_backup(self):
         server = self.create_test_server(wait_until='ACTIVE')
@@ -847,7 +853,6 @@
         cls.server_id = cls.recreate_server(None, volume_backed=True,
                                             validatable=True)
 
-    @utils.services('volume')
     @decorators.idempotent_id('6652dab9-ea24-4c93-ab5a-93d79c3041cf')
     def test_rebuild_volume_backed_server(self):
         """Test rebuilding a volume backed server"""
diff --git a/tempest/api/compute/servers/test_server_addresses.py b/tempest/api/compute/servers/test_server_addresses.py
index 5a3f5d0..978a9da 100644
--- a/tempest/api/compute/servers/test_server_addresses.py
+++ b/tempest/api/compute/servers/test_server_addresses.py
@@ -14,7 +14,6 @@
 #    under the License.
 
 from tempest.api.compute import base
-from tempest.common import utils
 from tempest.lib import decorators
 
 
@@ -35,7 +34,6 @@
 
     @decorators.attr(type='smoke')
     @decorators.idempotent_id('6eb718c0-02d9-4d5e-acd1-4e0c269cef39')
-    @utils.services('network')
     def test_list_server_addresses(self):
         """Test listing server address
 
@@ -52,7 +50,6 @@
 
     @decorators.attr(type='smoke')
     @decorators.idempotent_id('87bbc374-5538-4f64-b673-2b0e4443cc30')
-    @utils.services('network')
     def test_list_server_addresses_by_network(self):
         """Test listing server addresses filtered by network addresses
 
diff --git a/tempest/api/compute/servers/test_server_addresses_negative.py b/tempest/api/compute/servers/test_server_addresses_negative.py
index e7444d2..bb21594 100644
--- a/tempest/api/compute/servers/test_server_addresses_negative.py
+++ b/tempest/api/compute/servers/test_server_addresses_negative.py
@@ -14,7 +14,6 @@
 #    under the License.
 
 from tempest.api.compute import base
-from tempest.common import utils
 from tempest.lib import decorators
 from tempest.lib import exceptions as lib_exc
 
@@ -35,7 +34,6 @@
 
     @decorators.attr(type=['negative'])
     @decorators.idempotent_id('02c3f645-2d2e-4417-8525-68c0407d001b')
-    @utils.services('network')
     def test_list_server_addresses_invalid_server_id(self):
         """List addresses request should fail if server id not in system"""
         self.assertRaises(lib_exc.NotFound, self.client.list_addresses,
@@ -43,7 +41,6 @@
 
     @decorators.attr(type=['negative'])
     @decorators.idempotent_id('a2ab5144-78c0-4942-a0ed-cc8edccfd9ba')
-    @utils.services('network')
     def test_list_server_addresses_by_network_neg(self):
         """List addresses by network should fail if network name not valid"""
         self.assertRaises(lib_exc.NotFound,
diff --git a/tempest/api/compute/test_tenant_networks.py b/tempest/api/compute/test_tenant_networks.py
index 17f4b80..da28b9b 100644
--- a/tempest/api/compute/test_tenant_networks.py
+++ b/tempest/api/compute/test_tenant_networks.py
@@ -14,8 +14,11 @@
 
 from tempest.api.compute import base
 from tempest.common import utils
+from tempest import config
 from tempest.lib import decorators
 
+CONF = config.CONF
+
 
 class ComputeTenantNetworksTest(base.BaseV2ComputeTest):
     """Test compute tenant networks API with microversion less than 2.36"""
@@ -23,6 +26,14 @@
     max_microversion = '2.35'
 
     @classmethod
+    def skip_checks(cls):
+        super(ComputeTenantNetworksTest, cls).skip_checks()
+        if not CONF.service_available.neutron:
+            skip_msg = (
+                "%s skipped as Neutron is not available" % cls.__name__)
+            raise cls.skipException(skip_msg)
+
+    @classmethod
     def resource_setup(cls):
         super(ComputeTenantNetworksTest, cls).resource_setup()
         cls.client = cls.os_primary.tenant_networks_client
diff --git a/tempest/api/identity/v3/test_users.py b/tempest/api/identity/v3/test_users.py
index dc6dd4a..b95bd75 100644
--- a/tempest/api/identity/v3/test_users.py
+++ b/tempest/api/identity/v3/test_users.py
@@ -31,6 +31,12 @@
     """Test identity user password"""
 
     @classmethod
+    def skip_checks(cls):
+        super(IdentityV3UsersTest, cls).skip_checks()
+        if not CONF.identity_feature_enabled.security_compliance:
+            raise cls.skipException("Security compliance not available.")
+
+    @classmethod
     def resource_setup(cls):
         super(IdentityV3UsersTest, cls).resource_setup()
         cls.creds = cls.os_primary.credentials
@@ -77,8 +83,6 @@
         time.sleep(1)
         self.non_admin_users_client.auth_provider.set_auth()
 
-    @testtools.skipUnless(CONF.identity_feature_enabled.security_compliance,
-                          'Security compliance not available.')
     @decorators.idempotent_id('ad71bd23-12ad-426b-bb8b-195d2b635f27')
     @testtools.skipIf(CONF.identity_feature_enabled.immutable_user_source,
                       'Skipped because environment has an '
@@ -107,8 +111,6 @@
                           user_id=self.user_id,
                           password=old_pass)
 
-    @testtools.skipUnless(CONF.identity_feature_enabled.security_compliance,
-                          'Security compliance not available.')
     @decorators.idempotent_id('941784ee-5342-4571-959b-b80dd2cea516')
     @testtools.skipIf(CONF.identity_feature_enabled.immutable_user_source,
                       'Skipped because environment has an '
@@ -142,8 +144,6 @@
         # A different password can be set
         self._update_password(original_password=new_pass1, password=new_pass2)
 
-    @testtools.skipUnless(CONF.identity_feature_enabled.security_compliance,
-                          'Security compliance not available.')
     @decorators.idempotent_id('a7ad8bbf-2cff-4520-8c1d-96332e151658')
     def test_user_account_lockout(self):
         """Test locking out user account after failure attempts"""
diff --git a/tempest/api/image/v2/admin/test_image_task.py b/tempest/api/image/v2/admin/test_image_task.py
new file mode 100644
index 0000000..8cebdae
--- /dev/null
+++ b/tempest/api/image/v2/admin/test_image_task.py
@@ -0,0 +1,139 @@
+# Copyright 2023 Red Hat, Inc.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from tempest.api.image import base
+from tempest.common import waiters
+from tempest import config
+from tempest.lib.common.utils import data_utils
+from tempest.lib import decorators
+
+CONF = config.CONF
+
+
+class ImageTaskCreate(base.BaseV2ImageAdminTest):
+    """Test image task operations"""
+
+    @classmethod
+    def skip_checks(cls):
+        # TODO(msava): Add additional skipcheck with task conversion_format and
+        # glance ceph backend then will be available
+        # in tempest image service config options.
+        super(ImageTaskCreate, cls).skip_checks()
+        if not CONF.image.http_image:
+            skip_msg = ("%s skipped as http_image is not available " %
+                        cls.__name__)
+            raise cls.skipException(skip_msg)
+
+    @classmethod
+    def resource_setup(cls):
+        super(ImageTaskCreate, cls).resource_setup()
+
+    @staticmethod
+    def _prepare_image_tasks_param(type="import",
+                                   disk_format=['qcow2'],
+                                   image_from_format=['qcow2'],
+                                   image_location=CONF.image.http_image):
+        # TODO(msava): Need to add additional disk formats then
+        # task conversion_format and glance Ceph backend will be
+        # available in image service options
+        """Prepare image task params.
+        By default, will create task type 'import'
+
+        The same index is used for both params and creates a task
+        :param type Type of the task.
+        :param disk_format: Each format in the list is a different task.
+        :param image_from_format: Each format in the list is a different task.
+        :param image_location Location to import image from.
+        :return: A list with all task.
+        """
+        i = 0
+        tasks = list()
+        while i < len(disk_format):
+            image_name = data_utils.rand_name("task_image")
+            image_property = {"container_format": "bare",
+                              "disk_format": disk_format[0],
+                              "visibility": "public",
+                              "name": image_name
+                              }
+            task = {
+                "type": type,
+                "input": {
+                    "image_properties": image_property,
+                    "import_from_format": image_from_format[0],
+                    "import_from": image_location
+                }
+            }
+            tasks.append(task)
+            i += 1
+        return tasks
+
+    def _verify_disk_format(self, task_body):
+        expected_disk_format = \
+            task_body['input']['image_properties']['disk_format']
+        image_id = task_body['result']['image_id']
+        observed_disk_format = self.admin_client.show_image(
+            image_id)['disk_format']
+        # If glance backend storage is Ceph glance will convert
+        # image to raw format.
+        # TODO(msava): Need to change next lines once task conversion_format
+        # and glance ceph backend will be available in image service options
+        if observed_disk_format == 'raw':
+            return
+        self.assertEqual(observed_disk_format, expected_disk_format,
+                         message="Expected disk format not match ")
+
+    @decorators.idempotent_id('669d5387-0340-4abf-b62d-7cc89f539c8c')
+    def test_image_tasks_create(self):
+        """Test task type 'import' image """
+
+        # Prepare params for task type 'import'
+        tasks = self._prepare_image_tasks_param()
+
+        # Create task type 'import'
+        body = self.os_admin.tasks_client.create_task(**tasks[0])
+        task_id = body['id']
+        task_body = waiters.wait_for_tasks_status(self.os_admin.tasks_client,
+                                                  task_id, 'success')
+        self.addCleanup(self.admin_client.delete_image,
+                        task_body['result']['image_id'])
+        task_image_id = task_body['result']['image_id']
+        waiters.wait_for_image_status(self.client, task_image_id, 'active')
+        self._verify_disk_format(task_body)
+
+        # Verify disk format
+        image_body = self.client.show_image(task_image_id)
+        task_disk_format = \
+            task_body['input']['image_properties']['disk_format']
+        image_disk_format = image_body['disk_format']
+        self.assertEqual(
+            image_disk_format, task_disk_format,
+            message="Image Disc format %s not match to expected %s"
+                    % (image_disk_format, task_disk_format))
+
+    @decorators.idempotent_id("ad6450c6-7060-4ee7-a2d1-41c2604b446c")
+    @decorators.attr(type=['negative'])
+    def test_task_create_fake_image_location(self):
+        http_fake_url = ''.join(
+            ["http://", data_utils.rand_name('dummy-img-file'), ".qcow2"])
+        task = self._prepare_image_tasks_param(
+            image_from_format=['qcow2'],
+            disk_format=['qcow2'],
+            image_location=http_fake_url)
+        body = self.os_admin.tasks_client.create_task(**task[0])
+        task_observed = \
+            waiters.wait_for_tasks_status(self.os_admin.tasks_client,
+                                          body['id'], 'failure')
+        task_observed = task_observed['status']
+        self.assertEqual(task_observed, 'failure')
diff --git a/tempest/api/network/test_service_providers.py b/tempest/api/network/test_service_providers.py
index 5af5244..e203a2c 100644
--- a/tempest/api/network/test_service_providers.py
+++ b/tempest/api/network/test_service_providers.py
@@ -10,8 +10,6 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-import testtools
-
 from tempest.api.network import base
 from tempest.common import utils
 from tempest.lib import decorators
@@ -20,10 +18,14 @@
 class ServiceProvidersTest(base.BaseNetworkTest):
     """Test network service providers"""
 
+    @classmethod
+    def skip_checks(cls):
+        super(ServiceProvidersTest, cls).skip_checks()
+        if not utils.is_extension_enabled('service-type', 'network'):
+            skip_msg = ("service-type extension not enabled.")
+            raise cls.skipException(skip_msg)
+
     @decorators.idempotent_id('2cbbeea9-f010-40f6-8df5-4eaa0c918ea6')
-    @testtools.skipUnless(
-        utils.is_extension_enabled('service-type', 'network'),
-        'service-type extension not enabled.')
     def test_service_providers_list(self):
         """Test listing network service providers"""
         body = self.service_providers_client.list_service_providers()
diff --git a/tempest/api/volume/admin/test_encrypted_volumes_extend.py b/tempest/api/volume/admin/test_encrypted_volumes_extend.py
index e85a00d..4506389 100644
--- a/tempest/api/volume/admin/test_encrypted_volumes_extend.py
+++ b/tempest/api/volume/admin/test_encrypted_volumes_extend.py
@@ -14,7 +14,6 @@
 
 from tempest.api.volume import base
 from tempest.api.volume import test_volumes_extend as extend
-from tempest.common import utils
 from tempest import config
 from tempest.lib import decorators
 
@@ -25,23 +24,25 @@
                                          base.BaseVolumeAdminTest):
     """Tests extending the size of an attached encrypted volume."""
 
+    @classmethod
+    def skip_checks(cls):
+        super(EncryptedVolumesExtendAttachedTest, cls).skip_checks()
+        if not CONF.service_available.nova:
+            skip_msg = ("%s skipped as Nova is not available" % cls.__name__)
+            raise cls.skipException(skip_msg)
+        if not CONF.volume_feature_enabled.extend_attached_encrypted_volume:
+            raise cls.skipException(
+                "Attached encrypted volume extend is disabled.")
+
     @decorators.idempotent_id('e93243ec-7c37-4b5b-a099-ebf052c13216')
-    @testtools.skipUnless(
-        CONF.volume_feature_enabled.extend_attached_encrypted_volume,
-        "Attached encrypted volume extend is disabled.")
-    @utils.services('compute')
     def test_extend_attached_encrypted_volume_luksv1(self):
         """LUKs v1 decrypts and extends through libvirt."""
         volume = self.create_encrypted_volume(encryption_provider="luks")
         self._test_extend_attached_volume(volume)
 
     @decorators.idempotent_id('381a2a3a-b2f4-4631-a910-720881f2cc2f')
-    @testtools.skipUnless(
-        CONF.volume_feature_enabled.extend_attached_encrypted_volume,
-        "Attached encrypted volume extend is disabled.")
     @testtools.skipIf(CONF.volume.storage_protocol == 'ceph',
                       'Ceph only supports LUKSv2 if doing host attach.')
-    @utils.services('compute')
     def test_extend_attached_encrypted_volume_luksv2(self):
         """LUKs v2 decrypts and extends through os-brick."""
         volume = self.create_encrypted_volume(encryption_provider="luks2")
diff --git a/tempest/api/volume/base.py b/tempest/api/volume/base.py
index a31390a..ad8f573 100644
--- a/tempest/api/volume/base.py
+++ b/tempest/api/volume/base.py
@@ -42,6 +42,10 @@
         if not CONF.service_available.cinder:
             skip_msg = ("%s skipped as Cinder is not available" % cls.__name__)
             raise cls.skipException(skip_msg)
+        if cls.create_default_network and not CONF.service_available.neutron:
+            skip_msg = (
+                "%s skipped as Neutron is not available" % cls.__name__)
+            raise cls.skipException(skip_msg)
 
         api_version_utils.check_skip_with_microversion(
             cls.volume_min_microversion, cls.volume_max_microversion,
diff --git a/tempest/api/volume/test_volumes_extend.py b/tempest/api/volume/test_volumes_extend.py
index c5c94e1..c766db8 100644
--- a/tempest/api/volume/test_volumes_extend.py
+++ b/tempest/api/volume/test_volumes_extend.py
@@ -18,7 +18,6 @@
 import testtools
 
 from tempest.api.volume import base
-from tempest.common import utils
 from tempest.common import waiters
 from tempest import config
 from tempest.lib import decorators
@@ -181,10 +180,16 @@
 
 class VolumesExtendAttachedTest(BaseVolumesExtendAttachedTest):
 
+    @classmethod
+    def skip_checks(cls):
+        super(VolumesExtendAttachedTest, cls).skip_checks()
+        if not CONF.service_available.nova:
+            skip_msg = ("%s skipped as Nova is not available" % cls.__name__)
+            raise cls.skipException(skip_msg)
+        if not CONF.volume_feature_enabled.extend_attached_volume:
+            raise cls.skipException("Attached volume extend is disabled.")
+
     @decorators.idempotent_id('301f5a30-1c6f-4ea0-be1a-91fd28d44354')
-    @testtools.skipUnless(CONF.volume_feature_enabled.extend_attached_volume,
-                          "Attached volume extend is disabled.")
-    @utils.services('compute')
     def test_extend_attached_volume(self):
         volume = self.create_volume()
         self._test_extend_attached_volume(volume)
diff --git a/tempest/clients.py b/tempest/clients.py
index 1aa34d0..99e114c 100644
--- a/tempest/clients.py
+++ b/tempest/clients.py
@@ -97,6 +97,7 @@
                 self.image_v2.NamespacePropertiesClient()
             self.namespace_tags_client = self.image_v2.NamespaceTagsClient()
             self.image_versions_client = self.image_v2.VersionsClient()
+            self.tasks_client = self.image_v2.TaskClient()
             # NOTE(danms): If no alternate endpoint is configured,
             # this client will work the same as the base self.images_client.
             # If your test needs to know if these are different, check the
diff --git a/tempest/common/waiters.py b/tempest/common/waiters.py
index d88bc05..291f201 100644
--- a/tempest/common/waiters.py
+++ b/tempest/common/waiters.py
@@ -224,6 +224,24 @@
     raise lib_exc.TimeoutException(message)
 
 
+def wait_for_tasks_status(client, task_id, status):
+    start = int(time.time())
+    while int(time.time()) - start < client.build_timeout:
+        task = client.show_tasks(task_id)
+        if task['status'] == status:
+            return task
+        time.sleep(client.build_interval)
+    message = ('Task %(task_id)s tasks: '
+               'failed to reach %(status)s state within the required '
+               'time (%(timeout)s s).' % {'task_id': task_id,
+                                          'status': status,
+                                          'timeout': client.build_timeout})
+    caller = test_utils.find_test_caller()
+    if caller:
+        message = '(%s) %s' % (caller, message)
+    raise lib_exc.TimeoutException(message)
+
+
 def wait_for_image_imported_to_stores(client, image_id, stores=None):
     """Waits for an image to be imported to all requested stores.
 
diff --git a/tempest/lib/services/image/v2/__init__.py b/tempest/lib/services/image/v2/__init__.py
index a2f5bdc..5e303e3 100644
--- a/tempest/lib/services/image/v2/__init__.py
+++ b/tempest/lib/services/image/v2/__init__.py
@@ -27,9 +27,11 @@
 from tempest.lib.services.image.v2.resource_types_client import \
     ResourceTypesClient
 from tempest.lib.services.image.v2.schemas_client import SchemasClient
+from tempest.lib.services.image.v2.tasks_client import TaskClient
 from tempest.lib.services.image.v2.versions_client import VersionsClient
 
+
 __all__ = ['ImageMembersClient', 'ImagesClient', 'ImageCacheClient',
            'NamespaceObjectsClient', 'NamespacePropertiesClient',
            'NamespaceTagsClient', 'NamespacesClient', 'ResourceTypesClient',
-           'SchemasClient', 'VersionsClient']
+           'SchemasClient', 'TaskClient', 'VersionsClient']
diff --git a/tempest/lib/services/image/v2/tasks_client.py b/tempest/lib/services/image/v2/tasks_client.py
new file mode 100644
index 0000000..2cb33eb
--- /dev/null
+++ b/tempest/lib/services/image/v2/tasks_client.py
@@ -0,0 +1,70 @@
+# Copyright 2023 Red Hat, Inc.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+
+from urllib import parse as urllib
+
+from oslo_serialization import jsonutils as json
+
+from tempest.lib.common import rest_client
+
+CHUNKSIZE = 1024 * 64  # 64kB
+
+
+class TaskClient(rest_client.RestClient):
+    api_version = "v2"
+
+    def create_task(self, **kwargs):
+        """Create a task.
+
+        For a full list of available parameters, please refer to the official
+        API reference:
+        https://developer.openstack.org/api-ref/image/v2/#create-task
+        """
+        data = json.dumps(kwargs)
+        resp, body = self.post('tasks', data)
+        self.expected_success(201, resp.status)
+        body = json.loads(body)
+        return rest_client.ResponseBody(resp, body)
+
+    def list_tasks(self, **kwargs):
+        """List tasks.
+
+        For a full list of available parameters, please refer to the official
+        API reference:
+        https://developer.openstack.org/api-ref/image/v2/#list-tasks
+        """
+        url = 'tasks'
+
+        if kwargs:
+            url += '?%s' % urllib.urlencode(kwargs)
+
+        resp, body = self.get(url)
+        self.expected_success(200, resp.status)
+        body = json.loads(body)
+        return rest_client.ResponseBody(resp, body)
+
+    def show_tasks(self, task_id):
+        """Show task details.
+
+        For a full list of available parameters, please refer to the official
+        API reference:
+        https://docs.openstack.org/api-ref/image/v2/#show-task-details
+        """
+        url = 'tasks/%s' % task_id
+        resp, body = self.get(url)
+        self.expected_success(200, resp.status)
+        body = json.loads(body)
+        return rest_client.ResponseBody(resp, body)
diff --git a/tempest/scenario/test_server_advanced_ops.py b/tempest/scenario/test_server_advanced_ops.py
index 990b325..1c2246d 100644
--- a/tempest/scenario/test_server_advanced_ops.py
+++ b/tempest/scenario/test_server_advanced_ops.py
@@ -14,7 +14,6 @@
 #    under the License.
 
 from oslo_log import log as logging
-import testtools
 
 from tempest.common import utils
 from tempest.common import waiters
@@ -36,14 +35,21 @@
     """
 
     @classmethod
+    def skip_checks(cls):
+        super(TestServerAdvancedOps, cls).skip_checks()
+        if not CONF.service_available.nova:
+            skip_msg = ("%s skipped as Nova is not available" % cls.__name__)
+            raise cls.skipException(skip_msg)
+        if not CONF.compute_feature_enabled.suspend:
+            raise cls.skipException("Suspend is not available.")
+
+    @classmethod
     def setup_credentials(cls):
         cls.set_network_resources(network=True, subnet=True)
         super(TestServerAdvancedOps, cls).setup_credentials()
 
     @decorators.attr(type='slow')
     @decorators.idempotent_id('949da7d5-72c8-4808-8802-e3d70df98e2c')
-    @testtools.skipUnless(CONF.compute_feature_enabled.suspend,
-                          'Suspend is not available.')
     @utils.services('compute')
     def test_server_sequence_suspend_resume(self):
         # We create an instance for use in this test
diff --git a/tempest/tests/lib/services/image/v2/test_image_tasks_client.py b/tempest/tests/lib/services/image/v2/test_image_tasks_client.py
new file mode 100644
index 0000000..6e3b3b5
--- /dev/null
+++ b/tempest/tests/lib/services/image/v2/test_image_tasks_client.py
@@ -0,0 +1,86 @@
+# Copyright 2023 Red Hat, Inc.  All rights reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from tempest.lib.services.image.v2 import tasks_client
+from tempest.tests.lib import fake_auth_provider
+from tempest.tests.lib.services import base
+
+
+class TestImageTaskClient(base.BaseServiceTest):
+    def setUp(self):
+        super(TestImageTaskClient, self).setUp()
+        fake_auth = fake_auth_provider.FakeAuthProvider()
+        self.client = tasks_client.TaskClient(
+            fake_auth, 'image', 'regionOne')
+
+    def test_list_task(self):
+        fake_result = {
+
+            "first": "/v2/tasks",
+            "schema": "/v2/schemas/tasks",
+            "tasks": [
+                {
+                    "id": "08b7e1c8-3821-4f54-b3b8-d6655d178cdf",
+                    "owner": "fa6c8c1600f4444281658a23ee6da8e8",
+                    "schema": "/v2/schemas/task",
+                    "self": "/v2/tasks/08b7e1c8-3821-4f54-b3b8-d6655d178cdf",
+                    "status": "processing",
+                    "type": "import"
+                    },
+                {
+                    "id": "231c311d-3557-4e23-afc4-6d98af1419e7",
+                    "owner": "fa6c8c1600f4444281658a23ee6da8e8",
+                    "schema": "/v2/schemas/task",
+                    "self": "/v2/tasks/231c311d-3557-4e23-afc4-6d98af1419e7",
+                    "status": "processing",
+                    "type": "import"
+                    }
+                ]
+            }
+        self.check_service_client_function(
+            self.client.list_tasks,
+            'tempest.lib.common.rest_client.RestClient.get',
+            fake_result,
+            mock_args=['tasks'])
+
+    def test_create_task(self):
+        fake_result = {
+            "type": "import",
+            "input": {
+                "import_from":
+                "http://download.cirros-cloud.net/0.6.1/ \
+                    cirros-0.6.1-x86_64-disk.img",
+                "import_from_format": "qcow2",
+                "image_properties": {
+                    "disk_format": "qcow2",
+                    "container_format": "bare"
+                }
+            }
+            }
+        self.check_service_client_function(
+            self.client.create_task,
+            'tempest.lib.common.rest_client.RestClient.post',
+            fake_result,
+            status=201)
+
+    def test_show_task(self):
+        fake_result = {
+            "task_id": "08b7e1c8-3821-4f54-b3b8-d6655d178cdf"
+            }
+        self.check_service_client_function(
+            self.client.show_tasks,
+            'tempest.lib.common.rest_client.RestClient.get',
+            fake_result,
+            status=200,
+            task_id="e485aab9-0907-4973-921c-bb6da8a8fcf8")