diff --git a/scenario/test_volumes.py b/scenario/test_volumes.py
new file mode 100644
index 0000000..5527068
--- /dev/null
+++ b/scenario/test_volumes.py
@@ -0,0 +1,136 @@
+#    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 logging
+import six
+
+from cinderclient import exceptions as cinder_exceptions
+
+from heat_integrationtests.common import test
+
+LOG = logging.getLogger(__name__)
+
+
+class VolumeBackupRestoreIntegrationTest(test.HeatIntegrationTest):
+
+    def setUp(self):
+        super(VolumeBackupRestoreIntegrationTest, self).setUp()
+        self.client = self.orchestration_client
+        if self.conf.keypair_name:
+            self.keypair_name = self.conf.keypair_name
+        else:
+            self.keypair = self.create_keypair()
+            self.keypair_name = self.keypair.id
+        self.volume_description = 'A test volume description 123'
+        self.volume_size = self.conf.volume_size
+
+    def _cinder_verify(self, volume_id, expected_status='available'):
+        self.assertIsNotNone(volume_id)
+        volume = self.volume_client.volumes.get(volume_id)
+        self.assertIsNotNone(volume)
+        self.assertEqual(expected_status, volume.status)
+        self.assertEqual(self.volume_size, volume.size)
+        self.assertEqual(self.volume_description,
+                         volume.display_description)
+
+    def _outputs_verify(self, stack, expected_status='available'):
+        self.assertEqual(expected_status,
+                         self._stack_output(stack, 'status'))
+        self.assertEqual(six.text_type(self.volume_size),
+                         self._stack_output(stack, 'size'))
+        self.assertEqual(self.volume_description,
+                         self._stack_output(stack, 'display_description'))
+
+    def _create_stack(self, template_name, add_parameters={}):
+        # TODO(shardy): refactor this into a generic base-class helper
+        net = self._get_default_network()
+        stack_name = self._stack_rand_name()
+        template = self._load_template(__file__, template_name)
+        parameters = {'key_name': self.keypair_name,
+                      'instance_type': self.conf.instance_type,
+                      'image_id': self.conf.minimal_image_ref,
+                      'volume_description': self.volume_description,
+                      'timeout': self.conf.build_timeout,
+                      'network': net['id']}
+        parameters.update(add_parameters)
+        ret_stack = self.client.stacks.create(
+            stack_name=stack_name,
+            template=template,
+            parameters=parameters)
+        stack_id = ret_stack['stack']['id']
+        stack = self.client.stacks.get(stack_id)
+        self.assertIsNotNone(stack)
+        stack_identifier = '%s/%s' % (stack_name, stack.id)
+
+        self.addCleanup(self._stack_delete, stack_identifier)
+        self._wait_for_stack_status(stack_identifier, 'CREATE_COMPLETE')
+        return stack, stack_identifier
+
+    def test_cinder_volume_create_backup_restore(self):
+        """Ensure the 'Snapshot' deletion policy works.
+
+           This requires a more complex test, but it tests several aspects
+           of the heat cinder resources:
+           1. Create a volume, attach it to an instance, write some data to it
+           2. Delete the stack, with 'Snapshot' specified, creates a backup
+           3. Check the snapshot has created a volume backup
+           4. Create a new stack, where the volume is created from the backup
+           5. Verify the test data written in (1) is present in the new volume
+        """
+        stack, stack_identifier = self._create_stack(
+            template_name='test_volumes_delete_snapshot.yaml',
+            add_parameters={'volume_size': self.volume_size})
+
+        # Verify with cinder that the volume exists, with matching details
+        volume_id = self._stack_output(stack, 'volume_id')
+        self._cinder_verify(volume_id, expected_status='in-use')
+
+        # Verify the stack outputs are as expected
+        self._outputs_verify(stack, expected_status='in-use')
+
+        # Delete the stack and ensure a backup is created for volume_id
+        # but the volume itself is gone
+        self.client.stacks.delete(stack_identifier)
+        self._wait_for_stack_status(stack_identifier, 'DELETE_COMPLETE')
+        self.assertRaises(cinder_exceptions.NotFound,
+                          self.volume_client.volumes.get,
+                          volume_id)
+
+        backups = self.volume_client.backups.list()
+        self.assertIsNotNone(backups)
+        backups_filtered = [b for b in backups if b.volume_id == volume_id]
+        self.assertEqual(1, len(backups_filtered))
+        backup = backups_filtered[0]
+        self.addCleanup(self.volume_client.backups.delete, backup.id)
+
+        # Now, we create another stack where the volume is created from the
+        # backup created by the previous stack
+        stack2, stack_identifier2 = self._create_stack(
+            template_name='test_volumes_create_from_backup.yaml',
+            add_parameters={'backup_id': backup.id})
+
+        # Verify with cinder that the volume exists, with matching details
+        volume_id2 = self._stack_output(stack2, 'volume_id')
+        self._cinder_verify(volume_id2, expected_status='in-use')
+
+        # Verify the stack outputs are as expected
+        self._outputs_verify(stack2, expected_status='in-use')
+        testfile_data = self._stack_output(stack2, 'testfile_data')
+        self.assertEqual('{"instance1": "Volume Data:ateststring"}',
+                         testfile_data)
+
+        # Delete the stack and ensure the volume is gone
+        self.client.stacks.delete(stack_identifier2)
+        self._wait_for_stack_status(stack_identifier2, 'DELETE_COMPLETE')
+        self.assertRaises(cinder_exceptions.NotFound,
+                          self.volume_client.volumes.get,
+                          volume_id2)