Merge "Add function tests for event sinks"

heat/engine/service.py

@@ -794,6 +794,7 @@ class EngineService(service.Service):
             action = stack.CREATE
             if stack.adopt_stack_data:
                 action = stack.ADOPT
+            stack.thread_group_mgr = self.thread_group_mgr
             stack.converge_stack(template=stack.t, action=action)
         else:
             self.thread_group_mgr.start_with_lock(cnxt, stack, self.engine_id,
@@ -932,6 +933,7 @@ class EngineService(service.Service):
             cnxt, current_stack, template, params, files, args)
 
         if current_stack.convergence:
+            current_stack.thread_group_mgr = self.thread_group_mgr
             current_stack.converge_stack(template=tmpl,
                                          new_stack=updated_stack)
         else:
@@ -1284,6 +1286,7 @@ class EngineService(service.Service):
 
         if stack.convergence and cfg.CONF.convergence_engine:
             template = templatem.Template.create_empty_template()
+            stack.thread_group_mgr = self.thread_group_mgr
             stack.converge_stack(template=template, action=stack.DELETE)
             return
 

heat/engine/worker.py

@@ -305,6 +305,7 @@ class WorkerService(service.Service):
 
         tmpl = stack.t
         stack.adopt_stack_data = adopt_stack_data
+        stack.thread_group_mgr = self.thread_group_mgr
 
         if is_update:
             if (rsrc.replaced_by is not None and

heat/tests/convergence/framework/worker_wrapper.py

@@ -11,6 +11,8 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import mock
+
 from heat.engine import worker
 from heat.tests.convergence.framework import message_processor
 from heat.tests.convergence.framework import message_queue
@@ -28,7 +30,7 @@ class Worker(message_processor.MessageProcessor):
                        current_traversal, data,
                        is_update, adopt_stack_data):
         worker.WorkerService("fake_host", "fake_topic",
-                             "fake_engine", "tgm").check_resource(
+                             "fake_engine", mock.Mock()).check_resource(
                                  ctxt, resource_id,
                                  current_traversal,
                                  data, is_update,

heat_integrationtests/functional/test_event_sinks.py

@@ -0,0 +1,65 @@
+#    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 uuid
+
+from zaqarclient.queues.v1 import client as zaqarclient
+
+from heat_integrationtests.functional import functional_base
+
+
+class ZaqarEventSinkTest(functional_base.FunctionalTestsBase):
+    template = '''
+heat_template_version: "2013-05-23"
+resources:
+  test_resource:
+    type: OS::Heat::TestResource
+    properties:
+      value: ok
+'''
+
+    def test_events(self):
+        queue_id = str(uuid.uuid4())
+        environment = {'event_sinks': [{'type': 'zaqar-queue',
+                                        'target': queue_id,
+                                        'ttl': 120}]}
+        stack_identifier = self.stack_create(
+            template=self.template,
+            environment=environment)
+        stack_name, stack_id = stack_identifier.split('/')
+        conf = {
+            'auth_opts': {
+                'backend': 'keystone',
+                'options': {
+                    'os_username': self.conf.username,
+                    'os_password': self.conf.password,
+                    'os_project_name': self.conf.tenant_name,
+                    'os_auth_url': self.conf.auth_url
+                }
+            }
+        }
+
+        zaqar = zaqarclient.Client(conf=conf, version=1.1)
+        queue = zaqar.queue(queue_id)
+        messages = list(queue.messages())
+        self.assertEqual(4, len(messages))
+        types = [m.body['type'] for m in messages]
+        self.assertEqual(['os.heat.event'] * 4, types)
+        resources = set([m.body['payload']['resource_name'] for m in messages])
+        self.assertEqual(set([stack_name, 'test_resource']), resources)
+        stack_ids = [m.body['payload']['stack_id'] for m in messages]
+        self.assertEqual([stack_id] * 4, stack_ids)
+        statuses = [m.body['payload']['resource_status'] for m in messages]
+        self.assertEqual(
+            ['IN_PROGRESS', 'IN_PROGRESS', 'COMPLETE', 'COMPLETE'], statuses)
+        actions = [m.body['payload']['resource_action'] for m in messages]
+        self.assertEqual(['CREATE'] * 4, actions)