diff --git a/manila_tempest_tests/config.py b/manila_tempest_tests/config.py
index 9314da0..6d0a466 100644
--- a/manila_tempest_tests/config.py
+++ b/manila_tempest_tests/config.py
@@ -189,7 +189,7 @@
                      "careful enabling this opt."),
 
     cfg.StrOpt("image_with_share_tools",
-               default="manila-service-image",
+               default="manila-service-image-master",
                help="Image name for vm booting with nfs/smb clients tool."),
     cfg.StrOpt("image_username",
                default="manila",
diff --git a/manila_tempest_tests/tests/scenario/manager_share.py b/manila_tempest_tests/tests/scenario/manager_share.py
index 43ada92..01be412 100644
--- a/manila_tempest_tests/tests/scenario/manager_share.py
+++ b/manila_tempest_tests/tests/scenario/manager_share.py
@@ -97,6 +97,15 @@
         client.wait_for_share_status(share['id'], 'available')
         return share
 
+    def _create_snapshot(self, share_id, client=None, **kwargs):
+        client = client or self.shares_v2_client
+        snapshot = client.create_snapshot(share_id, **kwargs)
+        self.addCleanup(
+            client.wait_for_resource_deletion, snapshot_id=snapshot['id'])
+        self.addCleanup(client.delete_snapshot, snapshot['id'])
+        client.wait_for_snapshot_status(snapshot["id"], "available")
+        return snapshot
+
     def _wait_for_share_server_deletion(self, sn_id, client=None):
         """Wait for a share server to be deleted
 
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 10218a5..71ed0e3 100644
--- a/manila_tempest_tests/tests/scenario/test_share_basic_ops.py
+++ b/manila_tempest_tests/tests/scenario/test_share_basic_ops.py
@@ -49,9 +49,17 @@
     def setUp(self):
         super(ShareBasicOpsBase, self).setUp()
         base.verify_test_has_appropriate_tags(self)
+        self.image_ref = None
         # Setup image and flavor the test instance
         # Support both configured and injected values
-        self.image_ref = None
+        self.floatings = {}
+        if self.protocol not in CONF.share.enable_protocols:
+            message = "%s tests are disabled" % self.protocol
+            raise self.skipException(message)
+        if self.protocol not in CONF.share.enable_ip_rules_for_protocols:
+            message = ("%s tests for access rules other than IP are disabled" %
+                       self.protocol)
+            raise self.skipException(message)
         if not hasattr(self, 'flavor_ref'):
             self.flavor_ref = CONF.share.client_vm_flavor_ref
         if CONF.share.image_with_share_tools:
@@ -70,8 +78,7 @@
                       image=self.image_ref, flavor=self.flavor_ref,
                       ssh_user=self.ssh_user))
         self.security_group = self._create_security_group()
-        if CONF.share.multitenancy_enabled:
-            self.create_share_network()
+        self.create_share_network()
 
     def boot_instance(self, wait_until="ACTIVE"):
         self.keypair = self.create_keypair()
@@ -91,6 +98,7 @@
         # Obtain a floating IP
         floating_ip = (self.compute_floating_ips_client.create_floating_ip()
                        ['floating_ip'])
+        self.floatings[instance['id']] = floating_ip
         self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                         self.compute_floating_ips_client.delete_floating_ip,
                         floating_ip['id'])
@@ -112,11 +120,12 @@
             ssh_client.exec_command("ping -c 1 %s" % server_ip)
         return ssh_client
 
-    def mount_share(self, location, ssh_client):
+    def mount_share(self, location, ssh_client, target_dir=None):
         raise NotImplementedError
 
-    def umount_share(self, ssh_client):
-        ssh_client.exec_command("sudo umount /mnt")
+    def umount_share(self, ssh_client, target_dir=None):
+        target_dir = target_dir or "/mnt"
+        ssh_client.exec_command("sudo umount %s" % target_dir)
 
     def write_data(self, data, ssh_client):
         ssh_client.exec_command("echo \"%s\" | sudo tee /mnt/t1 && sudo sync" %
@@ -151,17 +160,20 @@
         return self._create_share_type(
             data_utils.rand_name("share_type"),
             extra_specs={
+                'snapshot_support': CONF.share.capability_snapshot_support,
                 'driver_handles_share_servers': CONF.share.multitenancy_enabled
             },)['share_type']
 
-    def create_share(self):
-        kwargs = {
+    def create_share(self, **kwargs):
+        kwargs.update({
             'share_protocol': self.protocol,
-            'share_type_id': self._get_share_type()['id'],
-        }
+        })
+        if not ('share_type_id' in kwargs or 'snapshot_id' in kwargs):
+            kwargs.update({'share_type_id': self._get_share_type()['id']})
         if CONF.share.multitenancy_enabled:
             kwargs.update({'share_network_id': self.share_net['id']})
         self.share = self._create_share(**kwargs)
+        return self.share
 
     def allow_access_ip(self, share_id, ip=None, instance=None, cleanup=True):
         if instance and not ip:
@@ -179,6 +191,18 @@
         self._allow_access(share_id, access_type='ip', access_to=ip,
                            cleanup=cleanup)
 
+    def provide_access_to_auxiliary_instance(self, instance, share=None):
+        share = share or self.share
+        if self.protocol.lower() == 'cifs':
+            self.allow_access_ip(share['id'], instance=instance, cleanup=False)
+        elif not CONF.share.multitenancy_enabled:
+            self.allow_access_ip(
+                share['id'], ip=self.floatings[instance['id']]['ip'],
+                instance=instance, cleanup=False)
+        elif (CONF.share.multitenancy_enabled and
+              self.protocol.lower() == 'nfs'):
+            self.allow_access_ip(share['id'], instance=instance, cleanup=False)
+
     def wait_for_active_instance(self, instance_id):
         waiters.wait_for_server_status(
             self.manager.servers_client, instance_id, "ACTIVE")
@@ -189,10 +213,10 @@
         instance = self.boot_instance(wait_until="BUILD")
         self.create_share()
         instance = self.wait_for_active_instance(instance["id"])
-        self.allow_access_ip(self.share['id'], instance=instance,
-                             cleanup=False)
         ssh_client = self.init_ssh(instance)
 
+        self.provide_access_to_auxiliary_instance(instance)
+
         if utils.is_microversion_lt(CONF.share.max_api_microversion, "2.9"):
             locations = self.share['export_locations']
         else:
@@ -217,9 +241,8 @@
         instance2 = self.wait_for_active_instance(instance2["id"])
 
         # Write data to first VM
-        self.allow_access_ip(self.share['id'], instance=instance1,
-                             cleanup=False)
         ssh_client_inst1 = self.init_ssh(instance1)
+        self.provide_access_to_auxiliary_instance(instance1)
 
         if utils.is_microversion_lt(CONF.share.max_api_microversion, "2.9"):
             locations = self.share['export_locations']
@@ -234,9 +257,8 @@
         self.write_data(test_data, ssh_client_inst1)
 
         # Read from second VM
-        self.allow_access_ip(
-            self.share['id'], instance=instance2, cleanup=False)
         ssh_client_inst2 = self.init_ssh(instance2)
+        self.provide_access_to_auxiliary_instance(instance2)
         self.mount_share(locations[0], ssh_client_inst2)
         self.addCleanup(self.umount_share,
                         ssh_client_inst2)
@@ -249,7 +271,7 @@
                           "Share migration tests are disabled.")
     def test_migration_files(self):
 
-        if self.protocol != "NFS":
+        if self.protocol != "nfs":
             raise self.skipException("Only NFS protocol supported "
                                      "at this moment.")
 
@@ -275,9 +297,8 @@
 
         dest_pool = dest_pool['name']
 
-        self.allow_access_ip(
-            self.share['id'], instance=instance, cleanup=False)
         ssh_client = self.init_ssh(instance)
+        self.provide_access_to_auxiliary_instance(instance)
 
         if utils.is_microversion_lt(CONF.share.max_api_microversion, "2.9"):
             exports = self.share['export_locations']
@@ -290,24 +311,24 @@
 
         self.mount_share(exports[0], ssh_client)
 
-        ssh_client.exec_command("mkdir -p /mnt/f1")
-        ssh_client.exec_command("mkdir -p /mnt/f2")
-        ssh_client.exec_command("mkdir -p /mnt/f3")
-        ssh_client.exec_command("mkdir -p /mnt/f4")
-        ssh_client.exec_command("mkdir -p /mnt/f1/ff1")
+        ssh_client.exec_command("sudo mkdir -p /mnt/f1")
+        ssh_client.exec_command("sudo mkdir -p /mnt/f2")
+        ssh_client.exec_command("sudo mkdir -p /mnt/f3")
+        ssh_client.exec_command("sudo mkdir -p /mnt/f4")
+        ssh_client.exec_command("sudo mkdir -p /mnt/f1/ff1")
         ssh_client.exec_command("sleep 1")
-        ssh_client.exec_command("dd if=/dev/zero of=/mnt/f1/1m1.bin bs=1M"
-                                " count=1")
-        ssh_client.exec_command("dd if=/dev/zero of=/mnt/f2/1m2.bin bs=1M"
-                                " count=1")
-        ssh_client.exec_command("dd if=/dev/zero of=/mnt/f3/1m3.bin bs=1M"
-                                " count=1")
-        ssh_client.exec_command("dd if=/dev/zero of=/mnt/f4/1m4.bin bs=1M"
-                                " count=1")
-        ssh_client.exec_command("dd if=/dev/zero of=/mnt/f1/ff1/1m5.bin bs=1M"
-                                " count=1")
-        ssh_client.exec_command("chmod -R 555 /mnt/f3")
-        ssh_client.exec_command("chmod -R 777 /mnt/f4")
+        ssh_client.exec_command(
+            "sudo dd if=/dev/zero of=/mnt/f1/1m1.bin bs=1M count=1")
+        ssh_client.exec_command(
+            "sudo dd if=/dev/zero of=/mnt/f2/1m2.bin bs=1M count=1")
+        ssh_client.exec_command(
+            "sudo dd if=/dev/zero of=/mnt/f3/1m3.bin bs=1M count=1")
+        ssh_client.exec_command(
+            "sudo dd if=/dev/zero of=/mnt/f4/1m4.bin bs=1M count=1")
+        ssh_client.exec_command(
+            "sudo dd if=/dev/zero of=/mnt/f1/ff1/1m5.bin bs=1M count=1")
+        ssh_client.exec_command("sudo chmod -R 555 /mnt/f3")
+        ssh_client.exec_command("sudo chmod -R 777 /mnt/f4")
 
         self.umount_share(ssh_client)
 
@@ -343,21 +364,114 @@
         self.assertIn('1m4.bin', output)
         self.assertIn('1m5.bin', output)
 
+    def _get_user_export_location(self, share):
+        if utils.is_microversion_lt(CONF.share.max_api_microversion, "2.9"):
+            user_export_location = share['export_locations'][0]
+        else:
+            exports = self.shares_v2_client.list_share_export_locations(
+                share['id'])
+            locations = [x['path'] for x in exports]
+            user_export_location = locations[0]
+        return user_export_location
+
+    @tc.attr(base.TAG_POSITIVE, base.TAG_BACKEND)
+    @testtools.skipUnless(
+        CONF.share.run_snapshot_tests, "Snapshot tests are disabled.")
+    def test_write_data_to_share_created_from_snapshot(self):
+        if self.protocol.upper() == 'CIFS':
+            msg = "Skipped for CIFS protocol because of bug/1649573"
+            raise self.skipException(msg)
+
+        # 1 - Create UVM, ok, created
+        instance = self.boot_instance(wait_until="BUILD")
+
+        # 2 - Create share S1, ok, created
+        parent_share = self.create_share()
+        instance = self.wait_for_active_instance(instance["id"])
+        self.addCleanup(self.servers_client.delete_server, instance['id'])
+
+        # 3 - SSH to UVM, ok, connected
+        ssh_client = self.init_ssh(instance)
+
+        # 4 - Provide RW access to S1, ok, provided
+        self.provide_access_to_auxiliary_instance(instance, parent_share)
+
+        # 5 - Try mount S1 to UVM, ok, mounted
+        user_export_location = self._get_user_export_location(parent_share)
+        parent_share_dir = "/mnt/parent"
+        ssh_client.exec_command("sudo mkdir -p %s" % parent_share_dir)
+        self.mount_share(user_export_location, ssh_client, parent_share_dir)
+        self.addCleanup(self.umount_share, ssh_client, parent_share_dir)
+
+        # 6 - Create "file1", ok, created
+        ssh_client.exec_command("sudo touch %s/file1" % parent_share_dir)
+
+        # 7 - Create snapshot SS1 from S1, ok, created
+        snapshot = self._create_snapshot(parent_share['id'])
+
+        # 8 - Create "file2" in share S1 - ok, created. We expect that
+        # snapshot will not contain any data created after snapshot creation.
+        ssh_client.exec_command("sudo touch %s/file2" % parent_share_dir)
+
+        # 9 - Create share S2 from SS1, ok, created
+        child_share = self.create_share(snapshot_id=snapshot["id"])
+
+        # 10 - Try mount S2 - fail, access denied. We test that child share
+        #      did not get access rules from parent share.
+        user_export_location = self._get_user_export_location(child_share)
+        child_share_dir = "/mnt/child"
+        ssh_client.exec_command("sudo mkdir -p %s" % child_share_dir)
+        self.assertRaises(
+            exceptions.SSHExecCommandFailed,
+            self.mount_share,
+            user_export_location, ssh_client, child_share_dir,
+        )
+
+        # 11 - Provide RW access to S2, ok, provided
+        self.provide_access_to_auxiliary_instance(instance, child_share)
+
+        # 12 - Try mount S2, ok, mounted
+        self.mount_share(user_export_location, ssh_client, child_share_dir)
+        self.addCleanup(self.umount_share, ssh_client, child_share_dir)
+
+        # 13 - List files on S2, only "file1" exists
+        output = ssh_client.exec_command("sudo ls -lRA %s" % child_share_dir)
+        self.assertIn('file1', output)
+        self.assertNotIn('file2', output)
+
+        # 14 - Create file3 on S2, ok, file created
+        ssh_client.exec_command("sudo touch %s/file3" % child_share_dir)
+
+        # 15 - List files on S1, two files exist - "file1" and "file2"
+        output = ssh_client.exec_command("sudo ls -lRA %s" % parent_share_dir)
+        self.assertIn('file1', output)
+        self.assertIn('file2', output)
+        self.assertNotIn('file3', output)
+
+        # 16 - List files on S2, two files exist - "file1" and "file3"
+        output = ssh_client.exec_command("sudo ls -lRA %s" % child_share_dir)
+        self.assertIn('file1', output)
+        self.assertNotIn('file2', output)
+        self.assertIn('file3', output)
+
 
 class TestShareBasicOpsNFS(ShareBasicOpsBase):
-    protocol = "NFS"
+    protocol = "nfs"
 
-    def mount_share(self, location, ssh_client):
-        ssh_client.exec_command("sudo mount -vt nfs \"%s\" /mnt" % location)
+    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 TestShareBasicOpsCIFS(ShareBasicOpsBase):
-    protocol = "CIFS"
+    protocol = "cifs"
 
-    def mount_share(self, location, ssh_client):
+    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\" /mnt -o guest" % location
+            "sudo mount.cifs \"%s\" %s -o guest" % (location, target_dir)
         )