heat_integrationtests.functional.test_event_sinks

Source code for heat_integrationtests.functional.test_event_sinks

#    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.common import test
from heat_integrationtests.functional import functional_base


[docs]class ZaqarEventSinkTest(functional_base.FunctionalTestsBase): template = ''' heat_template_version: "2013-05-23" resources: test_resource: type: OS::Heat::TestResource properties: value: ok '''
[docs] 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.project_name, 'os_auth_url': self.conf.auth_url, 'os_user_domain_id': self.conf.user_domain_id, 'os_project_domain_id': self.conf.project_domain_id, 'os_user_domain_name': self.conf.user_domain_name, 'os_project_domain_name': self.conf.project_domain_name } } } zaqar = zaqarclient.Client(conf=conf, version=1.1) queue = zaqar.queue(queue_id) def validate_messages(): messages = list(queue.messages()) if len(messages) < 4: return False 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] statuses.sort() self.assertEqual(['COMPLETE', 'COMPLETE', 'IN_PROGRESS', 'IN_PROGRESS'], statuses) actions = [m.body['payload']['resource_action'] for m in messages] self.assertEqual(['CREATE'] * 4, actions) return True self.assertTrue(test.call_until_true(20, 0, validate_messages))
Creative Commons Attribution 3.0 License

Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.