diff --git a/common/test.py b/common/test.py
index 30b10a8..b88c72b 100644
--- a/common/test.py
+++ b/common/test.py
@@ -275,7 +275,8 @@
             success_on_not_found=True)
 
     def update_stack(self, stack_identifier, template, environment=None,
-                     files=None, parameters=None):
+                     files=None, parameters=None,
+                     expected_status='UPDATE_COMPLETE'):
         env = environment or {}
         env_files = files or {}
         parameters = parameters or {}
@@ -289,9 +290,29 @@
             parameters=parameters,
             environment=env
         )
-        self._wait_for_stack_status(stack_identifier, 'UPDATE_COMPLETE')
+        self._wait_for_stack_status(stack_identifier, expected_status)
 
-    def assert_resource_is_a_stack(self, stack_identifier, res_name):
+    def assert_resource_is_a_stack(self, stack_identifier, res_name,
+                                   wait=False):
+        build_timeout = self.conf.build_timeout
+        build_interval = self.conf.build_interval
+        start = timeutils.utcnow()
+        while timeutils.delta_seconds(start,
+                                      timeutils.utcnow()) < build_timeout:
+            time.sleep(build_interval)
+            try:
+                nested_identifier = self._get_nested_identifier(
+                    stack_identifier, res_name)
+            except Exception:
+                # We may have to wait, if the create is in-progress
+                if wait:
+                    time.sleep(build_interval)
+                else:
+                    raise
+            else:
+                return nested_identifier
+
+    def _get_nested_identifier(self, stack_identifier, res_name):
         rsrc = self.client.resources.get(stack_identifier, res_name)
         nested_link = [l for l in rsrc.links if l['rel'] == 'nested']
         nested_href = nested_link[0]['href']
@@ -375,3 +396,22 @@
         stack_name = stack_identifier.split('/')[0]
         self.client.actions.resume(stack_name)
         self._wait_for_stack_status(stack_identifier, 'RESUME_COMPLETE')
+
+    def wait_for_event_with_reason(self, stack_identifier, reason,
+                                   rsrc_name=None, num_expected=1):
+        build_timeout = self.conf.build_timeout
+        build_interval = self.conf.build_interval
+        start = timeutils.utcnow()
+        while timeutils.delta_seconds(start,
+                                      timeutils.utcnow()) < build_timeout:
+            try:
+                rsrc_events = self.client.events.list(stack_identifier,
+                                                      resource_name=rsrc_name)
+            except heat_exceptions.HTTPNotFound:
+                LOG.debug("No events yet found for %s" % rsrc_name)
+            else:
+                matched = [e for e in rsrc_events
+                           if e.resource_status_reason == reason]
+                if len(matched) == num_expected:
+                    return matched
+            time.sleep(build_interval)
diff --git a/functional/test_hooks.py b/functional/test_hooks.py
new file mode 100644
index 0000000..f7d455a
--- /dev/null
+++ b/functional/test_hooks.py
@@ -0,0 +1,289 @@
+#    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 yaml
+
+from heat_integrationtests.common import test
+
+
+LOG = logging.getLogger(__name__)
+
+
+class HooksTest(test.HeatIntegrationTest):
+
+    def setUp(self):
+        super(HooksTest, self).setUp()
+        self.client = self.orchestration_client
+        self.template = {'heat_template_version': '2014-10-16',
+                         'resources': {
+                             'foo_step1': {'type': 'OS::Heat::RandomString'},
+                             'foo_step2': {'type': 'OS::Heat::RandomString',
+                                           'depends_on': 'foo_step1'},
+                             'foo_step3': {'type': 'OS::Heat::RandomString',
+                                           'depends_on': 'foo_step2'}}}
+
+    def test_hook_pre_create(self):
+        env = {'resource_registry':
+               {'resources':
+                {'foo_step2':
+                 {'hooks': 'pre-create'}}}}
+        # Note we don't wait for CREATE_COMPLETE, because we need to
+        # signal to clear the hook before create will complete
+        stack_identifier = self.stack_create(
+            template=self.template,
+            environment=env,
+            expected_status='CREATE_IN_PROGRESS')
+        self._wait_for_resource_status(
+            stack_identifier, 'foo_step1', 'CREATE_COMPLETE')
+        self._wait_for_resource_status(
+            stack_identifier, 'foo_step2', 'INIT_COMPLETE')
+        ev = self.wait_for_event_with_reason(
+            stack_identifier,
+            reason='CREATE paused until Hook pre-create is cleared',
+            rsrc_name='foo_step2')
+        self.assertEqual('INIT_COMPLETE', ev[0].resource_status)
+        self.client.resources.signal(stack_identifier, 'foo_step2',
+                                     data={'unset_hook': 'pre-create'})
+        ev = self.wait_for_event_with_reason(
+            stack_identifier,
+            reason='Hook pre-create is cleared',
+            rsrc_name='foo_step2')
+        self.assertEqual('INIT_COMPLETE', ev[0].resource_status)
+        self._wait_for_resource_status(
+            stack_identifier, 'foo_step2', 'CREATE_COMPLETE')
+        self._wait_for_stack_status(stack_identifier, 'CREATE_COMPLETE')
+
+    def test_hook_pre_update_nochange(self):
+        env = {'resource_registry':
+               {'resources':
+                {'foo_step2':
+                 {'hooks': 'pre-update'}}}}
+        stack_identifier = self.stack_create(
+            template=self.template,
+            environment=env)
+        res_before = self.client.resources.get(stack_identifier, 'foo_step2')
+        # Note we don't wait for UPDATE_COMPLETE, because we need to
+        # signal to clear the hook before update will complete
+        self.update_stack(
+            stack_identifier,
+            template=self.template,
+            environment=env,
+            expected_status='UPDATE_IN_PROGRESS')
+
+        # Note when a hook is specified, the resource status doesn't change
+        # when we hit the hook, so we look for the event, then assert the
+        # state is unchanged.
+        self._wait_for_resource_status(
+            stack_identifier, 'foo_step2', 'CREATE_COMPLETE')
+        ev = self.wait_for_event_with_reason(
+            stack_identifier,
+            reason='UPDATE paused until Hook pre-update is cleared',
+            rsrc_name='foo_step2')
+        self.assertEqual('CREATE_COMPLETE', ev[0].resource_status)
+        self.client.resources.signal(stack_identifier, 'foo_step2',
+                                     data={'unset_hook': 'pre-update'})
+        ev = self.wait_for_event_with_reason(
+            stack_identifier,
+            reason='Hook pre-update is cleared',
+            rsrc_name='foo_step2')
+        self.assertEqual('CREATE_COMPLETE', ev[0].resource_status)
+        self._wait_for_resource_status(
+            stack_identifier, 'foo_step2', 'CREATE_COMPLETE')
+        self._wait_for_stack_status(stack_identifier, 'UPDATE_COMPLETE')
+        res_after = self.client.resources.get(stack_identifier, 'foo_step2')
+        self.assertEqual(res_before.physical_resource_id,
+                         res_after.physical_resource_id)
+
+    def test_hook_pre_update_replace(self):
+        env = {'resource_registry':
+               {'resources':
+                {'foo_step2':
+                 {'hooks': 'pre-update'}}}}
+        stack_identifier = self.stack_create(
+            template=self.template,
+            environment=env)
+        res_before = self.client.resources.get(stack_identifier, 'foo_step2')
+        # Note we don't wait for UPDATE_COMPLETE, because we need to
+        # signal to clear the hook before update will complete
+        self.template['resources']['foo_step2']['properties'] = {'length': 10}
+        self.update_stack(
+            stack_identifier,
+            template=self.template,
+            environment=env,
+            expected_status='UPDATE_IN_PROGRESS')
+
+        # Note when a hook is specified, the resource status doesn't change
+        # when we hit the hook, so we look for the event, then assert the
+        # state is unchanged.
+        self._wait_for_resource_status(
+            stack_identifier, 'foo_step2', 'CREATE_COMPLETE')
+        ev = self.wait_for_event_with_reason(
+            stack_identifier,
+            reason='UPDATE paused until Hook pre-update is cleared',
+            rsrc_name='foo_step2')
+        self.assertEqual('CREATE_COMPLETE', ev[0].resource_status)
+        self.client.resources.signal(stack_identifier, 'foo_step2',
+                                     data={'unset_hook': 'pre-update'})
+        ev = self.wait_for_event_with_reason(
+            stack_identifier,
+            reason='Hook pre-update is cleared',
+            rsrc_name='foo_step2')
+        self.assertEqual('CREATE_COMPLETE', ev[0].resource_status)
+        self._wait_for_resource_status(
+            stack_identifier, 'foo_step2', 'CREATE_COMPLETE')
+        self._wait_for_stack_status(stack_identifier, 'UPDATE_COMPLETE')
+        res_after = self.client.resources.get(stack_identifier, 'foo_step2')
+        self.assertNotEqual(res_before.physical_resource_id,
+                            res_after.physical_resource_id)
+
+    def test_hook_pre_update_in_place(self):
+        env = {'resource_registry':
+               {'resources':
+                {'rg':
+                 {'hooks': 'pre-update'}}}}
+        template = {'heat_template_version': '2014-10-16',
+                    'resources': {
+                        'rg': {
+                            'type': 'OS::Heat::ResourceGroup',
+                            'properties': {
+                                'count': 1,
+                                'resource_def': {
+                                    'type': 'OS::Heat::RandomString'}}}}}
+        # Note we don't wait for CREATE_COMPLETE, because we need to
+        # signal to clear the hook before create will complete
+        stack_identifier = self.stack_create(
+            template=template,
+            environment=env)
+        res_before = self.client.resources.get(stack_identifier, 'rg')
+        template['resources']['rg']['properties']['count'] = 2
+        self.update_stack(
+            stack_identifier,
+            template=template,
+            environment=env,
+            expected_status='UPDATE_IN_PROGRESS')
+
+        # Note when a hook is specified, the resource status doesn't change
+        # when we hit the hook, so we look for the event, then assert the
+        # state is unchanged.
+        self._wait_for_resource_status(
+            stack_identifier, 'rg', 'CREATE_COMPLETE')
+        ev = self.wait_for_event_with_reason(
+            stack_identifier,
+            reason='UPDATE paused until Hook pre-update is cleared',
+            rsrc_name='rg')
+        self.assertEqual('CREATE_COMPLETE', ev[0].resource_status)
+        self.client.resources.signal(stack_identifier, 'rg',
+                                     data={'unset_hook': 'pre-update'})
+
+        ev = self.wait_for_event_with_reason(
+            stack_identifier,
+            reason='Hook pre-update is cleared',
+            rsrc_name='rg')
+        self.assertEqual('CREATE_COMPLETE', ev[0].resource_status)
+        self._wait_for_resource_status(
+            stack_identifier, 'rg', 'CREATE_COMPLETE')
+        self._wait_for_stack_status(stack_identifier, 'UPDATE_COMPLETE')
+        res_after = self.client.resources.get(stack_identifier, 'rg')
+        self.assertEqual(res_before.physical_resource_id,
+                         res_after.physical_resource_id)
+
+    def test_hook_pre_create_nested(self):
+        files = {'nested.yaml': yaml.dump(self.template)}
+        env = {'resource_registry':
+               {'resources':
+                {'nested':
+                 {'foo_step2':
+                  {'hooks': 'pre-create'}}}}}
+        template = {'heat_template_version': '2014-10-16',
+                    'resources': {
+                        'nested': {'type': 'nested.yaml'}}}
+        # Note we don't wait for CREATE_COMPLETE, because we need to
+        # signal to clear the hook before create will complete
+        stack_identifier = self.stack_create(
+            template=template,
+            environment=env,
+            files=files,
+            expected_status='CREATE_IN_PROGRESS')
+        self._wait_for_resource_status(stack_identifier, 'nested',
+                                       'CREATE_IN_PROGRESS')
+        nested_identifier = self.assert_resource_is_a_stack(
+            stack_identifier, 'nested', wait=True)
+        self._wait_for_resource_status(
+            nested_identifier, 'foo_step1', 'CREATE_COMPLETE')
+        self._wait_for_resource_status(
+            nested_identifier, 'foo_step2', 'INIT_COMPLETE')
+        ev = self.wait_for_event_with_reason(
+            nested_identifier,
+            reason='CREATE paused until Hook pre-create is cleared',
+            rsrc_name='foo_step2')
+        self.assertEqual('INIT_COMPLETE', ev[0].resource_status)
+        self.client.resources.signal(nested_identifier, 'foo_step2',
+                                     data={'unset_hook': 'pre-create'})
+        ev = self.wait_for_event_with_reason(
+            nested_identifier,
+            reason='Hook pre-create is cleared',
+            rsrc_name='foo_step2')
+        self.assertEqual('INIT_COMPLETE', ev[0].resource_status)
+        self._wait_for_resource_status(
+            nested_identifier, 'foo_step2', 'CREATE_COMPLETE')
+        self._wait_for_stack_status(stack_identifier, 'CREATE_COMPLETE')
+
+    def test_hook_pre_create_wildcard(self):
+        env = {'resource_registry':
+               {'resources':
+                {'foo_*':
+                 {'hooks': 'pre-create'}}}}
+        # Note we don't wait for CREATE_COMPLETE, because we need to
+        # signal to clear the hook before create will complete
+        stack_identifier = self.stack_create(
+            template=self.template,
+            environment=env,
+            expected_status='CREATE_IN_PROGRESS')
+        self._wait_for_resource_status(
+            stack_identifier, 'foo_step1', 'INIT_COMPLETE')
+        self.wait_for_event_with_reason(
+            stack_identifier,
+            reason='CREATE paused until Hook pre-create is cleared',
+            rsrc_name='foo_step1')
+        self.client.resources.signal(stack_identifier, 'foo_step1',
+                                     data={'unset_hook': 'pre-create'})
+        self.wait_for_event_with_reason(
+            stack_identifier,
+            reason='Hook pre-create is cleared',
+            rsrc_name='foo_step1')
+        self._wait_for_resource_status(
+            stack_identifier, 'foo_step2', 'INIT_COMPLETE')
+        self.wait_for_event_with_reason(
+            stack_identifier,
+            reason='CREATE paused until Hook pre-create is cleared',
+            rsrc_name='foo_step2')
+        self.client.resources.signal(stack_identifier, 'foo_step2',
+                                     data={'unset_hook': 'pre-create'})
+        self.wait_for_event_with_reason(
+            stack_identifier,
+            reason='Hook pre-create is cleared',
+            rsrc_name='foo_step2')
+        self._wait_for_resource_status(
+            stack_identifier, 'foo_step3', 'INIT_COMPLETE')
+        self.wait_for_event_with_reason(
+            stack_identifier,
+            reason='CREATE paused until Hook pre-create is cleared',
+            rsrc_name='foo_step3')
+        self.client.resources.signal(stack_identifier, 'foo_step3',
+                                     data={'unset_hook': 'pre-create'})
+        self.wait_for_event_with_reason(
+            stack_identifier,
+            reason='Hook pre-create is cleared',
+            rsrc_name='foo_step3')
+        self._wait_for_stack_status(stack_identifier, 'CREATE_COMPLETE')