diff --git a/.zuul.yaml b/.zuul.yaml
index 0e333c4..8738ae2 100644
--- a/.zuul.yaml
+++ b/.zuul.yaml
@@ -4,6 +4,7 @@
       - tempest-plugin-jobs
     check:
       jobs:
+        - cinder-tempest-plugin-lvm-multiattach
         - cinder-tempest-plugin-lvm-lio-barbican
         - cinder-tempest-plugin-lvm-lio-barbican-centos-9-stream:
             voting: false
@@ -55,6 +56,28 @@
         - cinder-tempest-plugin
 
 - job:
+    name: cinder-tempest-plugin-lvm-multiattach
+    description: |
+      This enables multiattach tests along with standard tempest tests
+    parent: devstack-tempest
+    required-projects:
+      - opendev.org/openstack/tempest
+      - opendev.org/openstack/cinder-tempest-plugin
+      - opendev.org/openstack/cinder
+    vars:
+      tempest_test_regex: '(^tempest\.(api|scenario)|(^cinder_tempest_plugin))'
+      tempest_test_exclude_list: '{{ ansible_user_dir }}/{{ zuul.projects["opendev.org/openstack/tempest"].src_dir }}/tools/tempest-integrated-gate-storage-exclude-list.txt'
+      tox_envlist: all
+      devstack_localrc:
+        ENABLE_VOLUME_MULTIATTACH: true
+      tempest_plugins:
+        - cinder-tempest-plugin
+    irrelevant-files:
+      - ^.*\.rst$
+      - ^doc/.*$
+      - ^releasenotes/.*$
+
+- job:
     name: cinder-tempest-plugin-lvm-barbican-base-abstract
     description: |
       This is a base job for lvm with lio & tgt targets
diff --git a/cinder_tempest_plugin/scenario/manager.py b/cinder_tempest_plugin/scenario/manager.py
index 862432c..8598ade 100644
--- a/cinder_tempest_plugin/scenario/manager.py
+++ b/cinder_tempest_plugin/scenario/manager.py
@@ -125,6 +125,40 @@
                                   server=instance)
         return count, md5_sum
 
+    def write_data_to_device(self, ip_address, out_dev, in_dev='/dev/urandom',
+                             bs=1024, count=100, private_key=None,
+                             server=None, sha_sum=False):
+        ssh_client = self.get_remote_client(
+            ip_address, private_key=private_key, server=server)
+
+        # Write data to device
+        write_command = (
+            'sudo dd bs=%(bs)s count=%(count)s if=%(in_dev)s of=%(out_dev)s '
+            '&& sudo dd bs=%(bs)s count=%(count)s if=%(out_dev)s' %
+            {'bs': str(bs), 'count': str(count), 'in_dev': in_dev,
+             'out_dev': out_dev})
+        if sha_sum:
+            # If we want to read sha1sum instead of the device data
+            write_command += ' | sha1sum | head -c 40'
+        data = ssh_client.exec_command(write_command)
+
+        return data
+
+    def read_data_from_device(self, ip_address, in_dev, bs=1024, count=100,
+                              private_key=None, server=None, sha_sum=False):
+        ssh_client = self.get_remote_client(
+            ip_address, private_key=private_key, server=server)
+
+        # Read data from device
+        read_command = ('sudo dd bs=%(bs)s count=%(count)s if=%(in_dev)s' %
+                        {'bs': bs, 'count': count, 'in_dev': in_dev})
+        if sha_sum:
+            # If we want to read sha1sum instead of the device data
+            read_command += ' | sha1sum  | head -c 40'
+        data = ssh_client.exec_command(read_command)
+
+        return data
+
     def _attach_and_get_volume_device_name(self, server, volume, instance_ip,
                                            private_key):
         ssh_client = self.get_remote_client(
diff --git a/cinder_tempest_plugin/scenario/test_volume_multiattach.py b/cinder_tempest_plugin/scenario/test_volume_multiattach.py
new file mode 100644
index 0000000..235cb25
--- /dev/null
+++ b/cinder_tempest_plugin/scenario/test_volume_multiattach.py
@@ -0,0 +1,136 @@
+# Copyright 2022 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 import config
+from tempest.lib import decorators
+from tempest.lib import exceptions as lib_exc
+
+from cinder_tempest_plugin.scenario import manager
+from tempest.scenario import manager as tempest_manager
+
+CONF = config.CONF
+
+
+class VolumeMultiattachTests(manager.ScenarioTest,
+                             tempest_manager.EncryptionScenarioTest):
+
+    compute_min_microversion = '2.60'
+    compute_max_microversion = 'latest'
+
+    def setUp(self):
+        super(VolumeMultiattachTests, self).setUp()
+        self.keypair = self.create_keypair()
+        self.security_group = self.create_security_group()
+
+    @classmethod
+    def skip_checks(cls):
+        super(VolumeMultiattachTests, cls).skip_checks()
+        if not CONF.compute_feature_enabled.volume_multiattach:
+            raise cls.skipException('Volume multi-attach is not available.')
+
+    def _verify_attachment(self, volume_id, server_id):
+        volume = self.volumes_client.show_volume(volume_id)['volume']
+        server_ids = (
+            [attachment['server_id'] for attachment in volume['attachments']])
+        self.assertIn(server_id, server_ids)
+
+    @decorators.idempotent_id('e6604b85-5280-4f7e-90b5-186248fd3423')
+    def test_multiattach_data_integrity(self):
+
+        # Create an instance
+        server_1 = self.create_server(
+            key_name=self.keypair['name'],
+            security_groups=[{'name': self.security_group['name']}])
+
+        # Create multiattach type
+        multiattach_vol_type = self.create_volume_type(
+            extra_specs={'multiattach': "<is> True"})
+
+        # Create a multiattach volume
+        volume = self.create_volume(volume_type=multiattach_vol_type['id'])
+
+        # Create encrypted volume
+        encrypted_volume = self.create_encrypted_volume(
+            'luks', volume_type='luks')
+
+        # Create a normal volume
+        simple_volume = self.create_volume()
+
+        # Attach normal and encrypted volumes (These volumes are not used in
+        # the current test but is used to emulate a real world scenario
+        # where different types of volumes will be attached to the server)
+        self.attach_volume(server_1, simple_volume)
+        self.attach_volume(server_1, encrypted_volume)
+
+        instance_ip = self.get_server_ip(server_1)
+
+        # Attach volume to instance and find it's device name (eg: /dev/vdb)
+        volume_device_name_inst_1, __ = (
+            self._attach_and_get_volume_device_name(
+                server_1, volume, instance_ip, self.keypair['private_key']))
+
+        out_device = '/dev/' + volume_device_name_inst_1
+
+        # This data is written from the first server and will be used to
+        # verify when reading data from second server
+        device_data_inst_1 = self.write_data_to_device(
+            instance_ip, out_device, private_key=self.keypair['private_key'],
+            server=server_1, sha_sum=True)
+
+        # Create another instance
+        server_2 = self.create_server(
+            key_name=self.keypair['name'],
+            security_groups=[{'name': self.security_group['name']}])
+
+        instance_2_ip = self.get_server_ip(server_2)
+
+        # Attach volume to instance and find it's device name (eg: /dev/vdc)
+        volume_device_name_inst_2, __ = (
+            self._attach_and_get_volume_device_name(
+                server_2, volume, instance_2_ip, self.keypair['private_key']))
+
+        in_device = '/dev/' + volume_device_name_inst_2
+
+        # Read data from volume device
+        device_data_inst_2 = self.read_data_from_device(
+            instance_2_ip, in_device, private_key=self.keypair['private_key'],
+            server=server_2, sha_sum=True)
+
+        self._verify_attachment(volume['id'], server_1['id'])
+        self._verify_attachment(volume['id'], server_2['id'])
+        self.assertEqual(device_data_inst_1, device_data_inst_2)
+
+    @decorators.idempotent_id('53514da8-f49c-4cda-8792-ff4a2fa69977')
+    def test_volume_multiattach_same_host_negative(self):
+        # Create an instance
+        server = self.create_server(
+            key_name=self.keypair['name'],
+            security_groups=[{'name': self.security_group['name']}])
+
+        # Create multiattach type
+        multiattach_vol_type = self.create_volume_type(
+            extra_specs={'multiattach': "<is> True"})
+
+        # Create an empty volume
+        volume = self.create_volume(volume_type=multiattach_vol_type['id'])
+
+        # Attach volume to instance
+        attachment = self.attach_volume(server, volume)
+
+        self.assertEqual(server['id'], attachment['serverId'])
+
+        # Try attaching the volume to the same instance
+        self.assertRaises(lib_exc.BadRequest, self.attach_volume, server,
+                          volume)