diff --git a/common/test.py b/common/test.py
index b88c72b..331c922 100644
--- a/common/test.py
+++ b/common/test.py
@@ -276,7 +276,8 @@
 
     def update_stack(self, stack_identifier, template, environment=None,
                      files=None, parameters=None,
-                     expected_status='UPDATE_COMPLETE'):
+                     expected_status='UPDATE_COMPLETE',
+                     disable_rollback=True):
         env = environment or {}
         env_files = files or {}
         parameters = parameters or {}
@@ -286,11 +287,19 @@
             stack_name=stack_name,
             template=template,
             files=env_files,
-            disable_rollback=True,
+            disable_rollback=disable_rollback,
             parameters=parameters,
             environment=env
         )
-        self._wait_for_stack_status(stack_identifier, expected_status)
+        kwargs = {'stack_identifier': stack_identifier,
+                  'status': expected_status}
+        if expected_status in ['ROLLBACK_COMPLETE']:
+            self.addCleanup(self.client.stacks.delete, stack_name)
+            # To trigger rollback you would intentionally fail the stack
+            # Hence check for rollback failures
+            kwargs['failure_pattern'] = '^ROLLBACK_FAILED$'
+
+        self._wait_for_stack_status(**kwargs)
 
     def assert_resource_is_a_stack(self, stack_identifier, res_name,
                                    wait=False):
@@ -334,7 +343,7 @@
 
     def stack_create(self, stack_name=None, template=None, files=None,
                      parameters=None, environment=None,
-                     expected_status='CREATE_COMPLETE'):
+                     expected_status='CREATE_COMPLETE', disable_rollback=True):
         name = stack_name or self._stack_rand_name()
         templ = template or self.template
         templ_files = files or {}
@@ -344,16 +353,23 @@
             stack_name=name,
             template=templ,
             files=templ_files,
-            disable_rollback=True,
+            disable_rollback=disable_rollback,
             parameters=params,
             environment=env
         )
-        self.addCleanup(self.client.stacks.delete, name)
+        if expected_status not in ['ROLLBACK_COMPLETE']:
+            self.addCleanup(self.client.stacks.delete, name)
 
         stack = self.client.stacks.get(name)
         stack_identifier = '%s/%s' % (name, stack.id)
+        kwargs = {'stack_identifier': stack_identifier,
+                  'status': expected_status}
         if expected_status:
-            self._wait_for_stack_status(stack_identifier, expected_status)
+            if expected_status in ['ROLLBACK_COMPLETE']:
+                # To trigger rollback you would intentionally fail the stack
+                # Hence check for rollback failures
+                kwargs['failure_pattern'] = '^ROLLBACK_FAILED$'
+            self._wait_for_stack_status(**kwargs)
         return stack_identifier
 
     def stack_adopt(self, stack_name=None, files=None,
diff --git a/common/test_resources/test_resource.py b/common/test_resources/test_resource.py
new file mode 100644
index 0000000..88f1745
--- /dev/null
+++ b/common/test_resources/test_resource.py
@@ -0,0 +1,134 @@
+#
+#    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 eventlet
+
+from heat.common.i18n import _
+from heat.engine import attributes
+from heat.engine import properties
+from heat.engine import resource
+from heat.engine import support
+from oslo_log import log as logging
+
+LOG = logging.getLogger(__name__)
+
+
+class TestResource(resource.Resource):
+    '''
+    A resource which stores the string value that was provided.
+
+    This resource is to be used only for testing.
+    It has control knobs such as 'update_replace', 'fail', 'wait_secs'
+
+    '''
+
+    support_status = support.SupportStatus(version='2014.1')
+
+    PROPERTIES = (
+        VALUE, UPDATE_REPLACE, FAIL, WAIT_SECS
+    ) = (
+        'value', 'update_replace', 'fail', 'wait_secs'
+    )
+
+    ATTRIBUTES = (
+        OUTPUT,
+    ) = (
+        'output',
+    )
+
+    properties_schema = {
+        VALUE: properties.Schema(
+            properties.Schema.STRING,
+            _('The input string to be stored.'),
+            default='test_string',
+            update_allowed=True
+        ),
+        FAIL: properties.Schema(
+            properties.Schema.BOOLEAN,
+            _('Value which can be set to fail the resource operation '
+              'to test failure scenarios.'),
+            update_allowed=True,
+            default=False
+        ),
+        UPDATE_REPLACE: properties.Schema(
+            properties.Schema.BOOLEAN,
+            _('Value which can be set to trigger update replace for '
+              'the particular resource'),
+            update_allowed=True,
+            default=False
+        ),
+        WAIT_SECS: properties.Schema(
+            properties.Schema.NUMBER,
+            _('Value which can be set for resource to wait after an action '
+              'is performed.'),
+            update_allowed=True,
+            default=0,
+        ),
+    }
+
+    attributes_schema = {
+        OUTPUT: attributes.Schema(
+            _('The string that was stored. This value is '
+              'also available by referencing the resource.'),
+            cache_mode=attributes.Schema.CACHE_NONE
+        ),
+    }
+
+    def handle_create(self):
+        value = self.properties.get(self.VALUE)
+        fail_prop = self.properties.get(self.FAIL)
+        sleep_secs = self.properties.get(self.WAIT_SECS)
+
+        self.data_set('value', value, redact=False)
+        self.resource_id_set(self.physical_resource_name())
+
+        # sleep for specified time
+        if sleep_secs:
+            LOG.debug("Resource %s sleeping for %s seconds",
+                      self.name, sleep_secs)
+            eventlet.sleep(sleep_secs)
+
+        # emulate failure
+        if fail_prop:
+            raise Exception("Test Resource failed %s", self.name)
+
+    def handle_update(self, json_snippet=None, tmpl_diff=None, prop_diff=None):
+        value = prop_diff.get(self.VALUE)
+        new_prop = json_snippet._properties
+        if value:
+            update_replace = new_prop.get(self.UPDATE_REPLACE, False)
+            if update_replace:
+                raise resource.UpdateReplace(self.name)
+            else:
+                fail_prop = new_prop.get(self.FAIL, False)
+                sleep_secs = new_prop.get(self.WAIT_SECS, 0)
+                # emulate failure
+                if fail_prop:
+                    raise Exception("Test Resource failed %s", self.name)
+                # update in place
+                self.data_set('value', value, redact=False)
+
+                if sleep_secs:
+                    LOG.debug("Update of Resource %s sleeping for %s seconds",
+                              self.name, sleep_secs)
+                    eventlet.sleep(sleep_secs)
+
+    def _resolve_attribute(self, name):
+        if name == self.OUTPUT:
+            return self.data().get('value')
+
+
+def resource_mapping():
+    return {
+        'OS::Heat::TestResource': TestResource,
+    }