diff --git a/manila_tempest_tests/config.py b/manila_tempest_tests/config.py
index 56a5b5a..0cd07d3 100644
--- a/manila_tempest_tests/config.py
+++ b/manila_tempest_tests/config.py
@@ -40,7 +40,7 @@
                     "This value is only used to validate the versions "
                     "response from Manila."),
     cfg.StrOpt("max_api_microversion",
-               default="2.74",
+               default="2.81",
                help="The maximum api microversion is configured to be the "
                     "value of the latest microversion supported by Manila."),
     cfg.StrOpt("region",
diff --git a/manila_tempest_tests/services/share/v2/json/shares_client.py b/manila_tempest_tests/services/share/v2/json/shares_client.py
index bf944d4..fe3e31c 100644
--- a/manila_tempest_tests/services/share/v2/json/shares_client.py
+++ b/manila_tempest_tests/services/share/v2/json/shares_client.py
@@ -2149,3 +2149,68 @@
         self.expected_success(200, resp.status)
         body = json.loads(body)
         return rest_client.ResponseBody(resp, body)
+
+#################
+
+    def create_resource_lock(self, resource_id, resource_type,
+                             resource_action='delete', lock_reason=None,
+                             version=LATEST_MICROVERSION):
+        body = {
+            "resource_lock": {
+                'resource_id': resource_id,
+                'resource_type': resource_type,
+                'resource_action': resource_action,
+                'lock_reason': lock_reason,
+            },
+        }
+        body = json.dumps(body)
+        resp, body = self.post("resource-locks", body, version=version)
+        self.expected_success(200, resp.status)
+        body = json.loads(body)
+        return rest_client.ResponseBody(resp, body)
+
+    def get_resource_lock(self, lock_id, version=LATEST_MICROVERSION):
+        resp, body = self.get("resource-locks/%s" % lock_id, version=version)
+
+        self.expected_success(200, resp.status)
+        body = json.loads(body)
+        return rest_client.ResponseBody(resp, body)
+
+    def list_resource_locks(self, filters=None, version=LATEST_MICROVERSION):
+        uri = (
+            "resource-locks?%s" % parse.urlencode(filters)
+            if filters else "resource-locks"
+        )
+
+        resp, body = self.get(uri, version=version)
+
+        self.expected_success(200, resp.status)
+        body = json.loads(body)
+        return rest_client.ResponseBody(resp, body)
+
+    def update_resource_lock(self,
+                             lock_id,
+                             resource_action=None,
+                             lock_reason=None,
+                             version=LATEST_MICROVERSION):
+        uri = 'resource-locks/%s' % lock_id
+        post_body = {}
+        if resource_action:
+            post_body['resource_action'] = resource_action
+        if lock_reason:
+            post_body['lock_reason'] = lock_reason
+        body = json.dumps({'resource_lock': post_body})
+
+        resp, body = self.put(uri, body, version=version)
+
+        self.expected_success(200, resp.status)
+        body = json.loads(body)
+        return rest_client.ResponseBody(resp, body)
+
+    def delete_resource_lock(self, lock_id, version=LATEST_MICROVERSION):
+        uri = "resource-locks/%s" % lock_id
+
+        resp, body = self.delete(uri, version=version)
+
+        self.expected_success(204, resp.status)
+        return rest_client.ResponseBody(resp, body)
diff --git a/manila_tempest_tests/tests/api/base.py b/manila_tempest_tests/tests/api/base.py
index 226e767..3cf70b4 100755
--- a/manila_tempest_tests/tests/api/base.py
+++ b/manila_tempest_tests/tests/api/base.py
@@ -792,6 +792,30 @@
         return security_service
 
     @classmethod
+    def create_resource_lock(cls, resource_id, resource_type='share',
+                             resource_action='delete', lock_reason=None,
+                             client=None, version=LATEST_MICROVERSION,
+                             cleanup_in_class=True):
+        lock_reason = lock_reason or "locked by tempest tests"
+        client = client or cls.shares_v2_client
+
+        lock = client.create_resource_lock(resource_id,
+                                           resource_type,
+                                           resource_action=resource_action,
+                                           lock_reason=lock_reason,
+                                           version=version)['resource_lock']
+        resource = {
+            "type": "resource_lock",
+            "id": lock["id"],
+            "client": client,
+        }
+        if cleanup_in_class:
+            cls.class_resources.insert(0, resource)
+        else:
+            cls.method_resources.insert(0, resource)
+        return lock
+
+    @classmethod
     def update_share_type(cls, share_type_id, name=None,
                           is_public=None, description=None,
                           client=None):
@@ -904,6 +928,8 @@
                     elif res["type"] == "quotas":
                         user_id = res.get('user_id')
                         client.reset_quotas(res_id, user_id=user_id)
+                    elif res["type"] == "resource_lock":
+                        client.delete_resource_lock(res_id)
                     else:
                         LOG.warning("Provided unsupported resource type for "
                                     "cleanup '%s'. Skipping.", res["type"])
diff --git a/manila_tempest_tests/tests/api/test_resource_locks.py b/manila_tempest_tests/tests/api/test_resource_locks.py
new file mode 100644
index 0000000..4f88d72
--- /dev/null
+++ b/manila_tempest_tests/tests/api/test_resource_locks.py
@@ -0,0 +1,291 @@
+# 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.
+
+import datetime
+
+from oslo_utils import timeutils
+from oslo_utils import uuidutils
+from tempest import config
+from tempest.lib import decorators
+from tempest.lib import exceptions as lib_exc
+
+from manila_tempest_tests.tests.api import base
+from manila_tempest_tests import utils
+
+CONF = config.CONF
+
+LOCKS_MIN_API_VERSION = '2.81'
+
+RESOURCE_LOCK_FIELDS = {
+    'id',
+    'resource_id',
+    'resource_action',
+    'resource_type',
+    'user_id',
+    'project_id',
+    'lock_context',
+    'created_at',
+    'updated_at',
+    'lock_reason',
+    'links',
+}
+
+
+class ResourceLockCRUTest(base.BaseSharesMixedTest):
+
+    @classmethod
+    def skip_checks(cls):
+        super(ResourceLockCRUTest, cls).skip_checks()
+        utils.check_skip_if_microversion_not_supported(LOCKS_MIN_API_VERSION)
+
+    @classmethod
+    def resource_setup(cls):
+        super(ResourceLockCRUTest, cls).resource_setup()
+        # create share type
+        share_type = cls.create_share_type()
+        cls.share_type_id = share_type['id']
+
+        # create share and place a "delete" lock on it
+        cls.share = cls.create_share(share_type_id=cls.share_type_id)
+        cls.lock = cls.create_resource_lock(cls.share['id'])
+
+    @decorators.attr(type=[base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND])
+    @decorators.idempotent_id('f3d162a6-2ab4-433b-b8e7-6bf4f0bb6b0e')
+    def test_list_resource_locks(self):
+        locks = self.shares_v2_client.list_resource_locks()['resource_locks']
+        self.assertIsInstance(locks, list)
+        self.assertIn(self.lock['id'], [x['id'] for x in locks])
+        lock = locks[0]
+        self.assertEqual(RESOURCE_LOCK_FIELDS, set(lock.keys()))
+
+    @decorators.attr(type=[base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND])
+    @decorators.idempotent_id('72cc0d43-f676-4dd8-8a93-faa71608de98')
+    def test_list_resource_locks_sorted_and_paginated(self):
+        lock_2 = self.create_resource_lock(self.share['id'],
+                                           cleanup_in_class=False)
+        lock_3 = self.create_resource_lock(self.share['id'],
+                                           cleanup_in_class=False)
+
+        expected_order = [self.lock['id'], lock_2['id']]
+
+        filters = {'sort_key': 'created_at', 'sort_dir': 'asc', 'limit': 2}
+        body = self.shares_v2_client.list_resource_locks(filters=filters)
+        # tempest/lib/common/rest_client.py's _parse_resp checks
+        # for number of keys in response's dict, if there is only single
+        # key, it returns directly this key, otherwise it returns
+        # parsed body. If limit param is used, then API returns
+        # multiple keys in response ('resource_locks' and
+        # 'resource_lock_links')
+        locks = body['resource_locks']
+        self.assertIsInstance(locks, list)
+        actual_order = [x['id'] for x in locks]
+        self.assertEqual(2, len(actual_order))
+        self.assertNotIn(lock_3['id'], actual_order)
+        self.assertEqual(expected_order, actual_order)
+
+    @decorators.attr(type=[base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND])
+    @decorators.idempotent_id('22831edc-9d99-432d-a0b6-85af8853db98')
+    def test_list_resource_locks_filtered(self):
+        # Filter by resource_id, resource_action, lock_reason_like,
+        # created_since, created_before
+        share_2 = self.create_share(share_type_id=self.share_type_id)
+        share_1_lock_2 = self.create_resource_lock(
+            self.share['id'],
+            lock_reason="clemson tigers rule",
+            cleanup_in_class=False)
+        share_2_lock = self.create_resource_lock(share_2['id'],
+                                                 cleanup_in_class=False)
+
+        # filter by resource_type
+        expected_locks = sorted([
+            self.lock['id'],
+            share_1_lock_2['id'],
+            share_2_lock['id']
+        ])
+        actual_locks = self.shares_v2_client.list_resource_locks(
+            filters={'resource_type': 'share'})['resource_locks']
+        self.assertEqual(expected_locks,
+                         sorted([lock['id'] for lock in actual_locks]))
+
+        # filter by resource_id
+        expected_locks = sorted([self.lock['id'], share_1_lock_2['id']])
+        actual_locks = self.shares_v2_client.list_resource_locks(
+            filters={'resource_id': self.share['id']})['resource_locks']
+        self.assertEqual(expected_locks,
+                         sorted([lock['id'] for lock in actual_locks]))
+
+        # filter by inexact lock reason
+        actual_locks = self.shares_v2_client.list_resource_locks(
+            filters={'lock_reason~': "clemson"})['resource_locks']
+        self.assertEqual([share_1_lock_2['id']],
+                         [lock['id'] for lock in actual_locks])
+
+        # timestamp filters
+        created_at_1 = timeutils.parse_strtime(self.lock['created_at'])
+        created_at_2 = timeutils.parse_strtime(share_2_lock['created_at'])
+        time_1 = created_at_1 - datetime.timedelta(seconds=1)
+        time_2 = created_at_2 - datetime.timedelta(microseconds=1)
+        filters_1 = {'created_since': str(time_1)}
+
+        # should return all resource locks created by this test including
+        # self.lock
+        actual_locks = self.shares_v2_client.list_resource_locks(
+            filters=filters_1)['resource_locks']
+        actual_lock_ids = [lock['id'] for lock in actual_locks]
+        self.assertGreaterEqual(len(actual_lock_ids), 3)
+        self.assertIn(self.lock['id'], actual_lock_ids)
+        self.assertIn(share_1_lock_2['id'], actual_lock_ids)
+
+        for lock in actual_locks:
+            time_diff_with_created_since = timeutils.delta_seconds(
+                time_1, timeutils.parse_strtime(lock['created_at']))
+            self.assertGreaterEqual(time_diff_with_created_since, 0)
+
+        filters_2 = {
+            'created_since': str(time_1),
+            'created_before': str(time_2),
+        }
+
+        actual_locks = self.shares_v2_client.list_resource_locks(
+            filters=filters_2)['resource_locks']
+        self.assertIsInstance(actual_locks, list)
+        actual_lock_ids = [lock['id'] for lock in actual_locks]
+        self.assertGreaterEqual(len(actual_lock_ids), 2)
+        self.assertIn(self.lock['id'], actual_lock_ids)
+        self.assertIn(share_1_lock_2['id'], actual_lock_ids)
+        self.assertNotIn(share_2_lock['id'], actual_lock_ids)
+
+    @decorators.attr(type=[base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND])
+    @decorators.idempotent_id('8cbf7331-f3a1-4c7b-ab1e-f8b938bf135e')
+    def test_get_resource_lock(self):
+        lock = self.shares_v2_client.get_resource_lock(
+            self.lock['id'])['resource_lock']
+
+        self.assertEqual(set(RESOURCE_LOCK_FIELDS), set(lock.keys()))
+        self.assertTrue(uuidutils.is_uuid_like(lock['id']))
+        self.assertEqual('share', lock['resource_type'])
+        self.assertEqual(self.share['id'], lock['resource_id'])
+        self.assertEqual('delete', lock['resource_action'])
+        self.assertEqual('user', lock['lock_context'])
+        self.assertEqual(self.shares_v2_client.user_id, lock['user_id'])
+        self.assertEqual(self.shares_v2_client.project_id, lock['project_id'])
+
+    @decorators.attr(type=[base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND])
+    @decorators.idempotent_id('a7f0fb6a-05ac-4afa-b8d9-04d20549bbd1')
+    def test_create_resource_lock(self):
+        # testing lock creation by a different user in the same project
+        project = self.os_admin.projects_client.show_project(
+            self.shares_v2_client.project_id)['project']
+        new_user_client = self.create_user_and_get_client(project)
+
+        lock = self.create_resource_lock(
+            self.share['id'],
+            client=new_user_client.shares_v2_client,
+            cleanup_in_class=False)
+
+        self.assertEqual(set(RESOURCE_LOCK_FIELDS), set(lock.keys()))
+        self.assertTrue(uuidutils.is_uuid_like(lock['id']))
+        self.assertEqual('share', lock['resource_type'])
+        self.assertEqual(self.share['id'], lock['resource_id'])
+        self.assertEqual('delete', lock['resource_action'])
+        self.assertEqual('user', lock['lock_context'])
+        self.assertEqual(new_user_client.shares_v2_client.user_id,
+                         lock['user_id'])
+        self.assertEqual(self.shares_v2_client.project_id, lock['project_id'])
+
+        # testing lock creation by admin
+        lock = self.create_resource_lock(
+            self.share['id'],
+            client=self.admin_shares_v2_client,
+            cleanup_in_class=False)
+        self.assertEqual('admin', lock['lock_context'])
+
+    @decorators.attr(type=[base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND])
+    @decorators.idempotent_id('d7b51cde-ff4f-45ce-a237-401e8be5b4e5')
+    def test_update_resource_lock(self):
+        lock = self.shares_v2_client.update_resource_lock(
+            self.lock['id'], lock_reason="new lock reason")['resource_lock']
+
+        # update is synchronous
+        self.assertEqual("new lock reason", lock['lock_reason'])
+
+        # verify get
+        lock = self.shares_v2_client.get_resource_lock(lock['id'])
+        self.assertEqual("new lock reason",
+                         lock['resource_lock']['lock_reason'])
+
+
+class ResourceLockDeleteTest(base.BaseSharesMixedTest):
+
+    @classmethod
+    def skip_checks(cls):
+        super(ResourceLockDeleteTest, cls).skip_checks()
+        utils.check_skip_if_microversion_not_supported(LOCKS_MIN_API_VERSION)
+
+    @classmethod
+    def resource_setup(cls):
+        super(ResourceLockDeleteTest, cls).resource_setup()
+        cls.share_type_id = cls.create_share_type()['id']
+
+    @decorators.idempotent_id('835fd617-4600-40a0-9ba1-40e5e0097b01')
+    @decorators.attr(type=[base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND])
+    def test_delete_lock(self):
+        share = self.create_share(share_type_id=self.share_type_id)
+        lock_1 = self.create_resource_lock(share['id'], cleanup_in_class=False)
+        lock_2 = self.create_resource_lock(share['id'], cleanup_in_class=False)
+
+        locks = self.shares_v2_client.list_resource_locks(
+            filters={'resource_id': share['id']})['resource_locks']
+        self.assertEqual(sorted([lock_1['id'], lock_2['id']]),
+                         sorted([lock['id'] for lock in locks]))
+
+        self.shares_v2_client.delete_resource_lock(lock_1['id'])
+        locks = self.shares_v2_client.list_resource_locks(
+            filters={'resource_id': share['id']})['resource_locks']
+        self.assertEqual(1, len(locks))
+        self.assertIn(lock_2['id'], [lock['id'] for lock in locks])
+
+    @decorators.idempotent_id('a96e70c7-0afe-4335-9abc-4b45ef778bd7')
+    @decorators.attr(type=[base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND])
+    def test_delete_locked_resource(self):
+        share = self.create_share(share_type_id=self.share_type_id)
+        lock_1 = self.create_resource_lock(share['id'], cleanup_in_class=False)
+        lock_2 = self.create_resource_lock(share['id'], cleanup_in_class=False)
+
+        # share can't be deleted when a lock exists
+        self.assertRaises(lib_exc.Forbidden,
+                          self.shares_v2_client.delete_share,
+                          share['id'])
+
+        # admin can't do this either
+        self.assertRaises(lib_exc.Forbidden,
+                          self.admin_shares_v2_client.delete_share,
+                          share['id'])
+        # "the force" shouldn't work either
+        self.assertRaises(lib_exc.Forbidden,
+                          self.admin_shares_v2_client.delete_share,
+                          share['id'],
+                          params={'force': True})
+
+        self.shares_v2_client.delete_resource_lock(lock_1['id'])
+
+        # there's at least one lock, share deletion should still fail
+        self.assertRaises(lib_exc.Forbidden,
+                          self.shares_v2_client.delete_share,
+                          share['id'])
+
+        self.shares_v2_client.delete_resource_lock(lock_2['id'])
+
+        # locks are gone, share deletion should be possible
+        self.shares_v2_client.delete_share(share['id'])
+        self.shares_v2_client.wait_for_resource_deletion(
+            share_id=share["id"])
diff --git a/manila_tempest_tests/tests/api/test_resource_locks_negative.py b/manila_tempest_tests/tests/api/test_resource_locks_negative.py
new file mode 100644
index 0000000..9501d6c
--- /dev/null
+++ b/manila_tempest_tests/tests/api/test_resource_locks_negative.py
@@ -0,0 +1,127 @@
+#    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 import config
+from tempest.lib import decorators
+from tempest.lib import exceptions as lib_exc
+
+from manila_tempest_tests.tests.api import base
+from manila_tempest_tests import utils
+
+CONF = config.CONF
+
+LOCKS_MIN_API_VERSION = '2.81'
+
+
+class ResourceLockNegativeTestAPIOnly(base.BaseSharesMixedTest):
+
+    @classmethod
+    def skip_checks(cls):
+        super(ResourceLockNegativeTestAPIOnly, cls).skip_checks()
+        utils.check_skip_if_microversion_not_supported(LOCKS_MIN_API_VERSION)
+
+    @decorators.attr(type=[base.TAG_NEGATIVE, base.TAG_API])
+    @decorators.idempotent_id('dd978cf7-1622-49e8-a6c8-3da4ac6c6f86')
+    def test_create_resource_lock_invalid_resource(self):
+        self.assertRaises(
+            lib_exc.BadRequest,
+            self.shares_v2_client.create_resource_lock,
+            'invalid-share-id',
+            'share'
+        )
+
+    @decorators.attr(type=[base.TAG_NEGATIVE, base.TAG_API])
+    @decorators.idempotent_id('d5600bdc-72c8-43fd-9900-c112aa6c87fa')
+    def test_delete_resource_lock_invalid(self):
+        self.assertRaises(
+            lib_exc.NotFound,
+            self.shares_v2_client.delete_resource_lock,
+            'invalid-lock-id'
+        )
+
+
+class ResourceLockNegativeTestWithShares(base.BaseSharesMixedTest):
+    @classmethod
+    def skip_checks(cls):
+        super(ResourceLockNegativeTestWithShares, cls).skip_checks()
+        utils.check_skip_if_microversion_not_supported(LOCKS_MIN_API_VERSION)
+
+    @classmethod
+    def resource_setup(cls):
+        super(ResourceLockNegativeTestWithShares, cls).resource_setup()
+        share_type = cls.create_share_type()
+        cls.share = cls.create_share(share_type_id=share_type['id'])
+        cls.user_project = cls.os_admin.projects_client.show_project(
+            cls.shares_v2_client.project_id)['project']
+
+    @decorators.attr(type=[base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND])
+    @decorators.idempotent_id('658297a8-d675-471d-8a19-3d9e9af3a352')
+    def test_create_resource_lock_invalid_resource_action(self):
+        self.assertRaises(
+            lib_exc.BadRequest,
+            self.shares_v2_client.create_resource_lock,
+            self.share['id'],
+            'share',
+            resource_action='invalid-action'
+        )
+
+    @decorators.attr(type=[base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND])
+    @decorators.idempotent_id('0057b3e7-c250-492d-805b-e355dff954ed')
+    def test_create_resource_lock_invalid_lock_reason_too_long(self):
+        self.assertRaises(
+            lib_exc.BadRequest,
+            self.shares_v2_client.create_resource_lock,
+            self.share['id'],
+            'share',
+            resource_action='delete',
+            lock_reason='invalid' * 150,
+        )
+
+    @decorators.attr(type=[base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND])
+    @decorators.idempotent_id('a2db3d29-b42f-4c0b-b484-afd32f91f747')
+    def test_update_resource_lock_invalid_param(self):
+        lock = self.create_resource_lock(self.share['id'])
+        self.assertRaises(
+            lib_exc.BadRequest,
+            self.shares_v2_client.update_resource_lock,
+            lock['id'],
+            resource_action='invalid-action'
+        )
+        self.assertRaises(
+            lib_exc.BadRequest,
+            self.shares_v2_client.update_resource_lock,
+            lock['id'],
+            lock_reason='invalid' * 150,
+        )
+
+    @decorators.attr(type=[base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND])
+    @decorators.idempotent_id('45b12120-0fc3-461f-8776-fdb92e599394')
+    def test_update_resource_lock_created_by_different_user(self):
+        lock = self.create_resource_lock(self.share['id'])
+        new_user = self.create_user_and_get_client(project=self.user_project)
+        self.assertRaises(
+            lib_exc.Forbidden,
+            new_user.shares_v2_client.update_resource_lock,
+            lock['id'],
+            lock_reason="I shouldn't be able to do this",
+        )
+
+    @decorators.attr(type=[base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND])
+    @decorators.idempotent_id('00a8ef2b-8769-4aad-aefc-43fc579492f7')
+    def test_delete_resource_lock_created_by_different_user(self):
+        lock = self.create_resource_lock(self.share['id'])
+        new_user = self.create_user_and_get_client(project=self.user_project)
+        self.assertRaises(
+            lib_exc.Forbidden,
+            new_user.shares_v2_client.delete_resource_lock,
+            lock['id'],
+        )