diff --git a/releasenotes/notes/add-location-api-5a57ab29dc6d6cd7.yaml b/releasenotes/notes/add-location-api-5a57ab29dc6d6cd7.yaml
new file mode 100644
index 0000000..f9166a2
--- /dev/null
+++ b/releasenotes/notes/add-location-api-5a57ab29dc6d6cd7.yaml
@@ -0,0 +1,4 @@
+---
+features:
+  - |
+    Add new location API support to image V2 client
diff --git a/tempest/api/compute/admin/test_live_migration.py b/tempest/api/compute/admin/test_live_migration.py
index f6a1ae9..fe8970f 100644
--- a/tempest/api/compute/admin/test_live_migration.py
+++ b/tempest/api/compute/admin/test_live_migration.py
@@ -34,6 +34,7 @@
 class LiveMigrationTestBase(base.BaseV2ComputeAdminTest):
     """Test live migration operations supported by admin user"""
 
+    credentials = ['primary', 'admin', 'project_manager']
     create_default_network = True
 
     @classmethod
@@ -51,13 +52,15 @@
     @classmethod
     def setup_clients(cls):
         super(LiveMigrationTestBase, cls).setup_clients()
-        cls.admin_migration_client = cls.os_admin.migrations_client
+        cls.migration_client = cls.os_admin.migrations_client
         cls.networks_client = cls.os_primary.networks_client
         cls.subnets_client = cls.os_primary.subnets_client
         cls.ports_client = cls.os_primary.ports_client
         cls.trunks_client = cls.os_primary.trunks_client
+        cls.mgr_server_client = cls.admin_servers_client
 
-    def _migrate_server_to(self, server_id, dest_host, volume_backed=False):
+    def _migrate_server_to(self, server_id, dest_host, volume_backed=False,
+                           use_manager_client=False):
         kwargs = dict()
         block_migration = getattr(self, 'block_migration', None)
         if self.block_migration is None:
@@ -66,7 +69,13 @@
             block_migration = (CONF.compute_feature_enabled.
                                block_migration_for_live_migration and
                                not volume_backed)
-        self.admin_servers_client.live_migrate_server(
+        if use_manager_client:
+            self.mgr_server_client = self.os_project_manager.servers_client
+            LOG.info("Using project manager for live migrating server: %s, "
+                     "project manager user id: %s",
+                     server_id, self.mgr_server_client.user_id)
+
+        self.mgr_server_client.live_migrate_server(
             server_id, host=dest_host, block_migration=block_migration,
             **kwargs)
 
@@ -74,11 +83,20 @@
                       volume_backed=False):
         # If target_host is None, check whether source host is different with
         # the new host after migration.
+        use_manager_client = False
         if target_host is None:
             source_host = self.get_host_for_server(server_id)
-        self._migrate_server_to(server_id, target_host, volume_backed)
+            # NOTE(gmaan): If new policy is enforced and and manager role
+            # is present in nova then use manager user to live migrate.
+            if (CONF.enforce_scope.nova and 'manager' in
+                CONF.compute_feature_enabled.nova_policy_roles):
+                use_manager_client = True
+
+        self._migrate_server_to(server_id, target_host, volume_backed,
+                                use_manager_client)
         waiters.wait_for_server_status(self.servers_client, server_id, state)
-        migration_list = (self.admin_migration_client.list_migrations()
+
+        migration_list = (self.os_admin.migrations_client.list_migrations()
                           ['migrations'])
 
         msg = ("Live Migration failed. Migrations list for Instance "
@@ -98,6 +116,9 @@
 class LiveMigrationTest(LiveMigrationTestBase):
     max_microversion = '2.24'
     block_migration = None
+    # If test case want to request the destination host to Nova
+    # otherwise Nova scheduler will pick one.
+    request_host = True
 
     @classmethod
     def setup_credentials(cls):
@@ -119,11 +140,12 @@
         server_id = self.create_test_server(wait_until="ACTIVE",
                                             volume_backed=volume_backed)['id']
         source_host = self.get_host_for_server(server_id)
-        if not CONF.compute_feature_enabled.can_migrate_between_any_hosts:
+        if (self.request_host and
+            CONF.compute_feature_enabled.can_migrate_between_any_hosts):
+            destination_host = self.get_host_other_than(server_id)
+        else:
             # not to specify a host so that the scheduler will pick one
             destination_host = None
-        else:
-            destination_host = self.get_host_other_than(server_id)
 
         if state == 'PAUSED':
             self.admin_servers_client.pause_server(server_id)
@@ -381,3 +403,16 @@
     min_microversion = '2.25'
     max_microversion = 'latest'
     block_migration = 'auto'
+
+
+class LiveMigrationWithoutHostTest(LiveMigrationTest):
+    # Test live migrations without host and let Nova scheduler will pick one.
+    request_host = False
+
+    @classmethod
+    def skip_checks(cls):
+        super(LiveMigrationWithoutHostTest, cls).skip_checks()
+        if not CONF.compute_feature_enabled.can_migrate_between_any_hosts:
+            skip_msg = ("Existing live migration tests are configured to live "
+                        "migrate without requesting host.")
+            raise cls.skipException(skip_msg)
diff --git a/tempest/api/compute/admin/test_migrations.py b/tempest/api/compute/admin/test_migrations.py
index fa8a737..88847e6 100644
--- a/tempest/api/compute/admin/test_migrations.py
+++ b/tempest/api/compute/admin/test_migrations.py
@@ -12,6 +12,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from oslo_log import log as logging
 import testtools
 
 from tempest.api.compute import base
@@ -22,15 +23,27 @@
 from tempest.lib import exceptions
 
 CONF = config.CONF
+LOG = logging.getLogger(__name__)
 
 
 class MigrationsAdminTest(base.BaseV2ComputeAdminTest):
     """Test migration operations supported by admin user"""
 
+    credentials = ['primary', 'admin', 'project_manager']
+
     @classmethod
     def setup_clients(cls):
         super(MigrationsAdminTest, cls).setup_clients()
         cls.client = cls.os_admin.migrations_client
+        cls.mgr_server_client = cls.admin_servers_client
+        # NOTE(gmaan): If new policy is enforced and and manager role
+        # is present in nova then use manager user to live migrate.
+        if (CONF.enforce_scope.nova and 'manager' in
+            CONF.compute_feature_enabled.nova_policy_roles):
+            cls.mgr_server_client = cls.os_project_manager.servers_client
+            LOG.info("Using project manager for migrating servers, "
+                     "project manager user id: %s",
+                     cls.mgr_server_client.user_id)
 
     @decorators.idempotent_id('75c0b83d-72a0-4cf8-a153-631e83e7d53f')
     def test_list_migrations(self):
@@ -143,7 +156,7 @@
         server = self.create_test_server(wait_until="ACTIVE")
         src_host = self.get_host_for_server(server['id'])
 
-        self.admin_servers_client.migrate_server(server['id'])
+        self.mgr_server_client.migrate_server(server['id'])
 
         waiters.wait_for_server_status(self.servers_client,
                                        server['id'], 'VERIFY_RESIZE')
diff --git a/tempest/api/image/v2/test_images.py b/tempest/api/image/v2/test_images.py
index 9309c76..4375da5 100644
--- a/tempest/api/image/v2/test_images.py
+++ b/tempest/api/image/v2/test_images.py
@@ -19,6 +19,7 @@
 
 from oslo_log import log as logging
 from tempest.api.image import base
+from tempest.common import image as image_utils
 from tempest.common import waiters
 from tempest import config
 from tempest.lib.common.utils import data_utils
@@ -980,3 +981,87 @@
         self.assertEqual(orig_image['os_hash_value'], image['os_hash_value'])
         self.assertEqual(orig_image['os_hash_algo'], image['os_hash_algo'])
         self.assertNotIn('validation_data', image['locations'][0])
+
+
+class HashCalculationRemoteDeletionTest(base.BaseV2ImageTest):
+    """Test calculation of image hash with new location API when the image is
+    deleted from a remote Glance service.
+    """
+    @classmethod
+    def resource_setup(cls):
+        super(HashCalculationRemoteDeletionTest,
+              cls).resource_setup()
+        if not cls.versions_client.has_version('2.17'):
+            # API is not new enough to support add location API
+            skip_msg = (
+                '%s skipped as Glance does not support v2.17')
+            raise cls.skipException(skip_msg)
+
+    @classmethod
+    def skip_checks(cls):
+        super(HashCalculationRemoteDeletionTest,
+              cls).skip_checks()
+        if not CONF.image_feature_enabled.do_secure_hash:
+            skip_msg = (
+                "%s skipped as do_secure_hash is disabled" %
+                cls.__name__)
+            raise cls.skipException(skip_msg)
+
+        if not CONF.image_feature_enabled.http_store_enabled:
+            skip_msg = (
+                "%s skipped as http store is disabled" %
+                cls.__name__)
+            raise cls.skipException(skip_msg)
+
+    @decorators.idempotent_id('123e4567-e89b-12d3-a456-426614174000')
+    def test_hash_calculation_cancelled(self):
+        """Test that image hash calculation is cancelled when the image
+        is deleted from a remote Glance service.
+
+        This test creates an image using new location API, verifies that
+        the hash calculation is initiated, and then deletes the image from a
+        remote Glance service, and verifies that the hash calculation process
+        is properly cancelled and image deleted successfully.
+        """
+
+        # Create an image with a location
+        image_name = data_utils.rand_name('image')
+        container_format = CONF.image.container_formats[0]
+        disk_format = CONF.image.disk_formats[0]
+        image = self.create_image(name=image_name,
+                                  container_format=container_format,
+                                  disk_format=disk_format,
+                                  visibility='private')
+        self.assertEqual(image_name, image['name'])
+        self.assertEqual('queued', image['status'])
+
+        # Start http server at random port to simulate the image location
+        # and to provide random data for the image with slow transfer
+        server = image_utils.RandomDataServer()
+        server.start()
+        self.addCleanup(server.stop)
+
+        # Add a location to the image
+        location = 'http://localhost:%d' % server.port
+        self.client.add_image_location(image['id'], location)
+        waiters.wait_for_image_status(self.client, image['id'], 'active')
+
+        # Verify that the hash calculation is initiated
+        image_info = self.client.show_image(image['id'])
+        self.assertEqual(CONF.image.hashing_algorithm,
+                         image_info['os_hash_algo'])
+        self.assertEqual('active', image_info['status'])
+
+        if CONF.image.alternate_image_endpoint:
+            # If alternate image endpoint is configured, we will delete the
+            # image from the alternate worker
+            self.os_primary.image_client_remote.delete_image(image['id'])
+        else:
+            # delete image from backend
+            self.client.delete_image(image['id'])
+
+        # If image is deleted successfully, the hash calculation is cancelled
+        self.client.wait_for_resource_deletion(image['id'])
+
+        # Stop the server to release the port
+        server.stop()
diff --git a/tempest/api/network/test_allowed_address_pair.py b/tempest/api/network/test_allowed_address_pair.py
index 01dda06..58160e0 100644
--- a/tempest/api/network/test_allowed_address_pair.py
+++ b/tempest/api/network/test_allowed_address_pair.py
@@ -124,7 +124,12 @@
     @decorators.idempotent_id('4d6d178f-34f6-4bff-a01c-0a2f8fe909e4')
     def test_update_port_with_cidr_address_pair(self):
         """Update allowed address pair with cidr"""
-        self._update_port_with_address(str(self.cidr))
+        # NOTE(slaweq): We need to use the next IP subnet to the one which
+        # is configured in the tempest config as the self.cidr will include
+        # "distributed" port created by the ML2/OVN backend and adding this
+        # particular IP address to the allowed address pair is forbidden by
+        # the ML2/OVN backend.
+        self._update_port_with_address(str(self.cidr.next()))
 
     @decorators.idempotent_id('b3f20091-6cd5-472b-8487-3516137df933')
     def test_update_port_with_multiple_ip_mac_address_pair(self):
diff --git a/tempest/api/object_storage/test_account_quotas.py b/tempest/api/object_storage/test_account_quotas.py
index 0a40237..37783b8 100644
--- a/tempest/api/object_storage/test_account_quotas.py
+++ b/tempest/api/object_storage/test_account_quotas.py
@@ -17,6 +17,7 @@
 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
 
@@ -52,7 +53,8 @@
             auth_data=self.reselleradmin_auth_data
         )
         # Set a quota of 20 bytes on the user's account before each test
-        headers = {"X-Account-Meta-Quota-Bytes": "20"}
+        self.set_quota = 20
+        headers = {"X-Account-Meta-Quota-Bytes": self.set_quota}
 
         self.os_roles_operator.account_client.request(
             "POST", url="", headers=headers, body="")
@@ -89,6 +91,24 @@
 
         self.assertHeaders(resp, 'Object', 'PUT')
 
+    @decorators.attr(type="smoke")
+    @decorators.idempotent_id('93fd7776-ae41-4949-8d0c-21889804c1ca')
+    @utils.requires_ext(extension='account_quotas', service='object')
+    def test_overlimit_upload(self):
+        """Test uploading an oversized object raises an OverLimit exception"""
+        object_name = data_utils.rand_name(
+            prefix=CONF.resource_name_prefix, name="TestObject")
+        data = data_utils.arbitrary_string(self.set_quota + 1)
+
+        nbefore = self._get_bytes_used()
+
+        self.assertRaises(lib_exc.OverLimit,
+                          self.object_client.create_object,
+                          self.container_name, object_name, data)
+
+        nafter = self._get_bytes_used()
+        self.assertEqual(nbefore, nafter)
+
     @decorators.attr(type=["smoke"])
     @decorators.idempotent_id('63f51f9f-5f1d-4fc6-b5be-d454d70949d6')
     @utils.requires_ext(extension='account_quotas', service='object')
@@ -115,3 +135,11 @@
 
             self.assertEqual(resp["status"], "204")
             self.assertHeaders(resp, 'Account', 'POST')
+
+    def _get_account_metadata(self):
+        resp, _ = self.account_client.list_account_metadata()
+        return resp
+
+    def _get_bytes_used(self):
+        resp = self._get_account_metadata()
+        return int(resp["x-account-bytes-used"])
diff --git a/tempest/api/volume/admin/test_encrypted_volumes_extend.py b/tempest/api/volume/admin/test_encrypted_volumes_extend.py
index 4506389..9c4819d 100644
--- a/tempest/api/volume/admin/test_encrypted_volumes_extend.py
+++ b/tempest/api/volume/admin/test_encrypted_volumes_extend.py
@@ -34,6 +34,7 @@
             raise cls.skipException(
                 "Attached encrypted volume extend is disabled.")
 
+    @decorators.skip_because(bug="2116852")
     @decorators.idempotent_id('e93243ec-7c37-4b5b-a099-ebf052c13216')
     def test_extend_attached_encrypted_volume_luksv1(self):
         """LUKs v1 decrypts and extends through libvirt."""
diff --git a/tempest/api/volume/test_volumes_actions.py b/tempest/api/volume/test_volumes_actions.py
index 8b2bc69..6261ddc 100644
--- a/tempest/api/volume/test_volumes_actions.py
+++ b/tempest/api/volume/test_volumes_actions.py
@@ -126,13 +126,6 @@
                             image_id)
             waiters.wait_for_image_status(self.images_client, image_id,
                                           'active')
-            # This is required for the optimized upload volume path.
-            # New location APIs are async so we need to wait for the location
-            # import task to complete.
-            # This should work with old location API since we don't fail if
-            # there are no tasks for the image
-            waiters.wait_for_image_tasks_status(self.images_client,
-                                                image_id, 'success')
             waiters.wait_for_volume_resource_status(self.volumes_client,
                                                     self.volume['id'],
                                                     'available')
diff --git a/tempest/common/image.py b/tempest/common/image.py
index 3618f7e..b8f76fb 100644
--- a/tempest/common/image.py
+++ b/tempest/common/image.py
@@ -14,6 +14,10 @@
 #    under the License.
 
 import copy
+from http import server
+import random
+import threading
+import time
 
 
 def get_image_meta_from_headers(resp):
@@ -63,3 +67,57 @@
         headers['x-image-meta-%s' % key] = str(value)
 
     return headers
+
+
+class RandomDataHandler(server.BaseHTTPRequestHandler):
+    def do_GET(self):
+        self.send_response(200)
+        self.send_header('Content-Type', 'application/octet-stream')
+        self.end_headers()
+
+        start_time = time.time()
+        chunk_size = 64 * 1024  # 64 KiB per chunk
+        while time.time() - start_time < 60:
+            data = bytes(random.getrandbits(8) for _ in range(chunk_size))
+            try:
+                self.wfile.write(data)
+                self.wfile.flush()
+                # simulate slow transfer
+                time.sleep(0.2)
+            except BrokenPipeError:
+                # Client disconnected; stop sending data
+                break
+
+    def do_HEAD(self):
+        # same size as in do_GET (19,660,800 bytes (about 18.75 MiB)
+        size = 300 * 65536
+        self.send_response(200)
+        self.send_header('Content-Type', 'application/octet-stream')
+        self.send_header('Content-Length', str(size))
+        self.end_headers()
+
+
+class RandomDataServer(object):
+    def __init__(self, handler_class=RandomDataHandler):
+        self.handler_class = handler_class
+        self.server = None
+        self.thread = None
+        self.port = None
+
+    def start(self):
+        # Bind to port 0 for an unused port
+        self.server = server.HTTPServer(('localhost', 0), self.handler_class)
+        self.port = self.server.server_address[1]
+
+        # Run server in background thread
+        self.thread = threading.Thread(target=self.server.serve_forever)
+        self.thread.daemon = True
+        self.thread.start()
+
+    def stop(self):
+        if self.server:
+            self.server.shutdown()
+            self.server.server_close()
+            self.thread.join()
+            self.server = None
+            self.thread = None
diff --git a/tempest/config.py b/tempest/config.py
index 5e5890b..fec7692 100644
--- a/tempest/config.py
+++ b/tempest/config.py
@@ -696,6 +696,11 @@
                          'vdi', 'iso', 'vhdx'],
                 help="A list of image's disk formats "
                      "users can specify."),
+    cfg.StrOpt('hashing_algorithm',
+               default='sha512',
+               help=('Hashing algorithm used by glance to calculate image '
+                     'hashes. This configuration value should be same as '
+                     'glance-api.conf: hashing_algorithm config option.')),
     cfg.StrOpt('images_manifest_file',
                default=None,
                help="A path to a manifest.yml generated using the "
@@ -740,6 +745,17 @@
                 help=('Indicates that image format is enforced by glance, '
                       'such that we should not expect to be able to upload '
                       'bad images for testing other services.')),
+    cfg.BoolOpt('do_secure_hash',
+                default=True,
+                help=('Is do_secure_hash enabled in glance. '
+                      'This configuration value should be same as '
+                      'glance-api.conf: do_secure_hash config option.')),
+    cfg.BoolOpt('http_store_enabled',
+                default=False,
+                help=('Is http store is enabled in glance. '
+                      'http store needs to be mentioned either in '
+                      'glance-api.conf: stores or in enabled_backends '
+                      'configuration option.')),
 ]
 
 network_group = cfg.OptGroup(name='network',
diff --git a/tempest/lib/services/image/v2/images_client.py b/tempest/lib/services/image/v2/images_client.py
index a6a1623..c491d9b 100644
--- a/tempest/lib/services/image/v2/images_client.py
+++ b/tempest/lib/services/image/v2/images_client.py
@@ -304,3 +304,13 @@
         resp, _ = self.delete(url)
         self.expected_success(204, resp.status)
         return rest_client.ResponseBody(resp)
+
+    def add_image_location(self, image_id, url, validation_data=None):
+        """Add location for specific Image."""
+        if not validation_data:
+            validation_data = {}
+        data = json.dumps({'url': url, 'validation_data': validation_data})
+        resp, _ = self.post('images/%s/locations' % (image_id),
+                            data)
+        self.expected_success(202, resp.status)
+        return rest_client.ResponseBody(resp)