diff --git a/manila_tempest_tests/common/constants.py b/manila_tempest_tests/common/constants.py
index 54d32fb..8791837 100644
--- a/manila_tempest_tests/common/constants.py
+++ b/manila_tempest_tests/common/constants.py
@@ -19,6 +19,7 @@
 STATUS_MIGRATING_TO = 'migrating_to'
 STATUS_CREATING = 'creating'
 STATUS_DELETING = 'deleting'
+STATUS_SHRINKING = 'shrinking'
 
 TEMPEST_MANILA_PREFIX = 'tempest-manila'
 
diff --git a/manila_tempest_tests/tests/scenario/manager_share.py b/manila_tempest_tests/tests/scenario/manager_share.py
index 1893107..5a38079 100644
--- a/manila_tempest_tests/tests/scenario/manager_share.py
+++ b/manila_tempest_tests/tests/scenario/manager_share.py
@@ -21,6 +21,7 @@
 from manila_tempest_tests.common import remote_client
 from manila_tempest_tests.tests.api import base
 from manila_tempest_tests.tests.scenario import manager
+from manila_tempest_tests import utils
 
 from tempest.common import waiters
 from tempest import config
@@ -201,7 +202,8 @@
         """
 
         remote_client.exec_command(
-            "sudo sh -c \"dd bs={} count={} if={} of={} conv=fsync\""
+            "sudo sh -c \"dd bs={} count={} if={} of={} conv=fsync"
+            " iflag=fullblock\""
             .format(block_size, block_count, input_file, output_file))
 
     def read_data_from_mounted_share(self,
@@ -347,6 +349,15 @@
                 'driver_handles_share_servers': CONF.share.multitenancy_enabled
             },)['share_type']
 
+    def get_share_export_locations(self, share):
+        if utils.is_microversion_lt(CONF.share.max_api_microversion, "2.9"):
+            locations = share['export_locations']
+        else:
+            exports = self.shares_v2_client.list_share_export_locations(
+                share['id'])
+            locations = [x['path'] for x in exports]
+        return locations
+
     def _get_ipv6_server_ip(self, instance):
         for net_list in instance['addresses'].values():
             for net_data in net_list:
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 4daa2f3..1e23b82 100644
--- a/manila_tempest_tests/tests/scenario/test_share_basic_ops.py
+++ b/manila_tempest_tests/tests/scenario/test_share_basic_ops.py
@@ -82,7 +82,7 @@
                                    error_on_invalid_ip_version=False):
         locations = None
         if share:
-            locations = self._get_share_export_locations(share)
+            locations = self.get_share_export_locations(share)
         elif snapshot:
             locations = self._get_snapshot_export_locations(snapshot)
 
@@ -93,17 +93,6 @@
 
         return locations
 
-    def _get_share_export_locations(self, share):
-
-        if utils.is_microversion_lt(CONF.share.max_api_microversion, "2.9"):
-            locations = share['export_locations']
-        else:
-            exports = self.shares_v2_client.list_share_export_locations(
-                share['id'])
-            locations = [x['path'] for x in exports]
-
-        return locations
-
     def _get_snapshot_export_locations(self, snapshot):
         exports = (self.shares_v2_client.
                    list_snapshot_export_locations(snapshot['id']))
diff --git a/manila_tempest_tests/tests/scenario/test_share_shrink.py b/manila_tempest_tests/tests/scenario/test_share_shrink.py
new file mode 100644
index 0000000..8bbb921
--- /dev/null
+++ b/manila_tempest_tests/tests/scenario/test_share_shrink.py
@@ -0,0 +1,177 @@
+#    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 six
+import time
+
+from oslo_log import log as logging
+from tempest import config
+from tempest.lib import exceptions
+import testtools
+from testtools import testcase as tc
+
+from manila_tempest_tests.common import constants
+from manila_tempest_tests.tests.api import base
+from manila_tempest_tests.tests.scenario import manager_share as manager
+
+CONF = config.CONF
+LOG = logging.getLogger(__name__)
+
+
+class ShareShrinkBase(manager.ShareScenarioTest):
+
+    """This test case uses the following flow:
+
+     * Launch an instance
+     * Create share (Configured size + 1)
+     * Configure RW access to the share
+     * Perform ssh to instance
+     * Mount share
+     * Write data in share (in excess of 1GB)
+     * Shrink share to 1GB (fail expected)
+     * Delete data from share
+     * Shrink share to 1GB
+     * Write more than 1GB of data (fail expected)
+     * Unmount share
+     * Delete share
+     * Terminate the instance
+    """
+
+    @tc.attr(base.TAG_POSITIVE, base.TAG_BACKEND)
+    @testtools.skipUnless(
+        CONF.share.run_shrink_tests, 'Shrink share tests are disabled.')
+    def test_create_shrink_and_write(self):
+        default_share_size = CONF.share.share_size
+        share_size = CONF.share.share_size + 1
+
+        LOG.debug('Step 1 - create instance')
+        instance = self.boot_instance(wait_until="BUILD")
+
+        LOG.debug('Step 2 - create share of size {} Gb'.format(share_size))
+        share = self.create_share(size=share_size)
+
+        LOG.debug('Step 3 - wait for active instance')
+        instance = self.wait_for_active_instance(instance["id"])
+        remote_client = self.init_remote_client(instance)
+
+        LOG.debug('Step 4 - grant access')
+        self.provide_access_to_auxiliary_instance(instance)
+
+        locations = self.get_share_export_locations(share)
+
+        LOG.debug('Step 5 - mount')
+        self.mount_share(locations[0], remote_client)
+
+        total_blocks = (1024 * default_share_size) / 64
+        blocks = total_blocks + 4
+        LOG.debug('Step 6 - writing {} * 64MB blocks'.format(blocks))
+        self.write_data_to_mounted_share_using_dd(remote_client,
+                                                  '/mnt/t1', '64M',
+                                                  blocks, '/dev/urandom')
+        ls_result = remote_client.exec_command("sudo ls -lAh /mnt/")
+        LOG.debug(ls_result)
+
+        LOG.debug('Step 8 - try update size, shrink and wait')
+        self.shares_v2_client.shrink_share(share['id'],
+                                           new_size=default_share_size)
+        self.shares_v2_client.wait_for_share_status(
+            share['id'], 'shrinking_possible_data_loss_error')
+
+        LOG.debug('Step 9 - delete data')
+        remote_client.exec_command("sudo rm /mnt/t1")
+
+        ls_result = remote_client.exec_command("sudo ls -lAh /mnt/")
+        LOG.debug(ls_result)
+
+        LOG.debug('Step 10 - reset and shrink')
+        self.share_shrink_retry_until_success(share["id"],
+                                              share_size=default_share_size)
+
+        share = self.shares_v2_client.get_share(share["id"])
+        self.assertEqual(default_share_size, int(share["size"]))
+
+        LOG.debug('Step 11 - write more data than allocated, should fail')
+        self.assertRaises(
+            exceptions.SSHExecCommandFailed,
+            self.write_data_to_mounted_share_using_dd,
+            remote_client, '/mnt/t1', '64M', blocks, '/dev/urandom')
+
+        LOG.debug('Step 12 - unmount')
+        self.unmount_share(remote_client)
+
+    def share_shrink_retry_until_success(self, share_id, share_size,
+                                         status_attr='status'):
+        """Try share reset, followed by shrink, until timeout"""
+
+        check_interval = CONF.share.build_interval * 2
+        body = self.shares_v2_client.get_share(share_id)
+        share_status = body[status_attr]
+        start = int(time.time())
+
+        while share_status != constants.STATUS_AVAILABLE:
+            if share_status != constants.STATUS_SHRINKING:
+                self.shares_admin_v2_client.reset_state(
+                    share_id, status=constants.STATUS_AVAILABLE)
+                try:
+                    self.shares_v2_client.shrink_share(share_id,
+                                                       new_size=share_size)
+                except exceptions.BadRequest as e:
+                    if ('New size for shrink must be less '
+                       'than current size') in six.text_type(e):
+                        break
+            time.sleep(check_interval)
+            body = self.shares_v2_client.get_share(share_id)
+            share_status = body[status_attr]
+            if share_status == constants.STATUS_AVAILABLE:
+                return
+
+            if int(time.time()) - start >= CONF.share.build_timeout:
+                message = ("Share's %(status_attr)s failed to transition to "
+                           "%(status)s within the required time %(seconds)s." %
+                           {"status_attr": status_attr,
+                            "status": constants.STATUS_AVAILABLE,
+                            "seconds": CONF.share.build_timeout})
+                raise exceptions.TimeoutException(message)
+
+
+class TestShareShrinkNFS(ShareShrinkBase):
+    protocol = "nfs"
+
+    def mount_share(self, location, ssh_client, target_dir=None):
+        target_dir = target_dir or "/mnt"
+        ssh_client.exec_command(
+            "sudo mount -vt nfs \"%s\" %s" % (location, target_dir)
+        )
+
+
+class TestShareShrinkCIFS(ShareShrinkBase):
+    protocol = "cifs"
+
+    def mount_share(self, location, ssh_client, target_dir=None):
+        location = location.replace("\\", "/")
+        target_dir = target_dir or "/mnt"
+        ssh_client.exec_command(
+            "sudo mount.cifs \"%s\" %s -o guest" % (location, target_dir)
+        )
+
+
+# NOTE(u_glide): this function is required to exclude ShareShrinkBase from
+# executed test cases.
+# See: https://docs.python.org/2/library/unittest.html#load-tests-protocol
+# for details.
+def load_tests(loader, tests, _):
+    result = []
+    for test_case in tests:
+        if type(test_case._tests[0]) is ShareShrinkBase:
+            continue
+        result.append(test_case)
+    return loader.suiteClass(result)