Use Zaqar for software-config transport

Add a new Zaqar transport to distribute metadata to servers and retrieve
data out of them.

blueprint software-config-zaqar
Depends-On: Icd30e32a17247790dfd57d6420d67f9140020db6
Change-Id: If23ffb3553c3506b981c06bfb82dae41b93590a5

heat/engine/clients/os/zaqar.py

@@ -27,6 +27,8 @@ class ZaqarClientPlugin(client_plugin.ClientPlugin):
 
     exceptions_module = zaqar_errors
 
+    DEFAULT_TTL = 3600
+
     def _create(self):
         return self.create_for_tenant(self.context.tenant_id)
 

heat/engine/resources/openstack/heat/software_deployment.py

@@ -97,20 +97,24 @@ class SoftwareDeployment(signal_responder.SignalResponder):
         DEPLOY_RESOURCE_NAME, DEPLOY_AUTH_URL,
         DEPLOY_USERNAME, DEPLOY_PASSWORD,
         DEPLOY_PROJECT_ID, DEPLOY_USER_ID,
-        DEPLOY_SIGNAL_VERB, DEPLOY_SIGNAL_TRANSPORT
+        DEPLOY_SIGNAL_VERB, DEPLOY_SIGNAL_TRANSPORT,
+        DEPLOY_QUEUE_ID
     ) = (
         'deploy_server_id', 'deploy_action',
         'deploy_signal_id', 'deploy_stack_id',
         'deploy_resource_name', 'deploy_auth_url',
         'deploy_username', 'deploy_password',
         'deploy_project_id', 'deploy_user_id',
-        'deploy_signal_verb', 'deploy_signal_transport'
+        'deploy_signal_verb', 'deploy_signal_transport',
+        'deploy_queue_id'
     )
 
     SIGNAL_TRANSPORTS = (
-        CFN_SIGNAL, TEMP_URL_SIGNAL, HEAT_SIGNAL, NO_SIGNAL
+        CFN_SIGNAL, TEMP_URL_SIGNAL, HEAT_SIGNAL, NO_SIGNAL,
+        ZAQAR_SIGNAL
     ) = (
-        'CFN_SIGNAL', 'TEMP_URL_SIGNAL', 'HEAT_SIGNAL', 'NO_SIGNAL'
+        'CFN_SIGNAL', 'TEMP_URL_SIGNAL', 'HEAT_SIGNAL', 'NO_SIGNAL',
+        'ZAQAR_SIGNAL'
     )
 
     properties_schema = {
@@ -194,6 +198,10 @@ class SoftwareDeployment(signal_responder.SignalResponder):
         return self.properties[
             self.SIGNAL_TRANSPORT] == self.TEMP_URL_SIGNAL
 
+    def _signal_transport_zaqar(self):
+        return self.properties.get(
+            self.SIGNAL_TRANSPORT) == self.ZAQAR_SIGNAL
+
     def _build_properties(self, properties, config_id, action):
         props = {
             'config_id': config_id,
@@ -321,6 +329,17 @@ class SoftwareDeployment(signal_responder.SignalResponder):
             container, object_name, '')
         return put_url
 
+    def _get_queue_id(self):
+        queue_id = self.data().get('signal_queue_id')
+        if queue_id:
+            return queue_id
+
+        queue_id = self.physical_resource_name()
+        zaqar = self.client('zaqar')
+        zaqar.queue(queue_id).ensure_exists()
+        self.data_set('signal_queue_id', queue_id)
+        return queue_id
+
     def _delete_temp_url(self):
         object_name = self.data().get('signal_object_name')
         if not object_name:
@@ -337,6 +356,17 @@ class SoftwareDeployment(signal_responder.SignalResponder):
         self.data_delete('signal_object_name')
         self.data_delete('signal_temp_url')
 
+    def _delete_queue(self):
+        queue_id = self.data().get('signal_queue_id')
+        if not queue_id:
+            return
+        zaqar = self.client('zaqar')
+        try:
+            zaqar.queue(queue_id).delete()
+        except Exception as ex:
+            self.client_plugin('zaqar').ignore_not_found(ex)
+        self.data_delete('signal_queue_id')
+
     def _build_derived_inputs(self, action, source):
         scl = sc.SoftwareConfig
         inputs = copy.deepcopy(source.get(scl.INPUTS)) or []
@@ -414,7 +444,7 @@ class SoftwareDeployment(signal_responder.SignalResponder):
                 scl.TYPE: 'String',
                 'value': 'PUT'
             })
-        elif self._signal_transport_heat():
+        elif self._signal_transport_heat() or self._signal_transport_zaqar():
             inputs.extend([{
                 scl.NAME: self.DEPLOY_AUTH_URL,
                 scl.DESCRIPTION: _('URL for API authentication'),
@@ -441,6 +471,14 @@ class SoftwareDeployment(signal_responder.SignalResponder):
                 scl.TYPE: 'String',
                 'value': self.stack.stack_user_project_id
             }])
+        if self._signal_transport_zaqar():
+            inputs.append({
+                scl.NAME: self.DEPLOY_QUEUE_ID,
+                scl.DESCRIPTION: _('ID of queue to use for signaling '
+                                   'output values'),
+                scl.TYPE: 'String',
+                'value': self._get_queue_id()
+            })
 
         return inputs
 
@@ -448,7 +486,7 @@ class SoftwareDeployment(signal_responder.SignalResponder):
         if self._signal_transport_cfn():
             self._create_user()
             self._create_keypair()
-        if self._signal_transport_heat():
+        if self._signal_transport_heat() or self._signal_transport_zaqar():
             self.password = uuid.uuid4().hex
             self._create_user()
         return self._handle_action(self.CREATE)
@@ -500,6 +538,8 @@ class SoftwareDeployment(signal_responder.SignalResponder):
             self._delete_user()
         elif self._signal_transport_temp_url():
             self._delete_temp_url()
+        elif self._signal_transport_zaqar():
+            self._delete_queue()
 
         derived_config_id = None
         if self.resource_id is not None:

heat/engine/resources/openstack/nova/server.py

@@ -109,9 +109,9 @@ class Server(stack_user.StackUser):
     )
 
     _SOFTWARE_CONFIG_TRANSPORTS = (
-        POLL_SERVER_CFN, POLL_SERVER_HEAT, POLL_TEMP_URL
+        POLL_SERVER_CFN, POLL_SERVER_HEAT, POLL_TEMP_URL, ZAQAR_MESSAGE
     ) = (
-        'POLL_SERVER_CFN', 'POLL_SERVER_HEAT', 'POLL_TEMP_URL'
+        'POLL_SERVER_CFN', 'POLL_SERVER_HEAT', 'POLL_TEMP_URL', 'ZAQAR_MESSAGE'
     )
 
     ATTRIBUTES = (
@@ -530,6 +530,21 @@ class Server(stack_user.StackUser):
                 'stack_id': self.stack.identifier().stack_path(),
                 'resource_name': self.name}
             }
+        if self.transport_zaqar_message():
+            queue_id = self.physical_resource_name()
+            self.data_set('metadata_queue_id', queue_id)
+            zaqar_plugin = self.client_plugin('zaqar')
+            zaqar = zaqar_plugin.create_for_tenant(
+                self.stack.stack_user_project_id)
+            queue = zaqar.queue(queue_id)
+            queue.post({'body': meta, 'ttl': zaqar_plugin.DEFAULT_TTL})
+            meta['os-collect-config'] = {'zaqar': {
+                'user_id': self._get_user_id(),
+                'password': self.password,
+                'auth_url': self.context.auth_url,
+                'project_id': self.stack.stack_user_project_id,
+                'queue_id': queue_id}
+            }
         elif self.transport_poll_server_cfn():
             meta['os-collect-config'] = {'cfn': {
                 'metadata_url': '%s/v1/' % cfg.CONF.heat_metadata_server_url,
@@ -577,7 +592,8 @@ class Server(stack_user.StackUser):
             self._create_user()
             self._create_keypair()
 
-        elif self.transport_poll_server_heat():
+        elif (self.transport_poll_server_heat() or
+              self.transport_zaqar_message()):
             self.password = uuid.uuid4().hex
             self._create_user()
 
@@ -621,6 +637,10 @@ class Server(stack_user.StackUser):
         return self.properties[
             self.SOFTWARE_CONFIG_TRANSPORT] == self.POLL_TEMP_URL
 
+    def transport_zaqar_message(self):
+        return self.properties.get(
+            self.SOFTWARE_CONFIG_TRANSPORT) == self.ZAQAR_MESSAGE
+
     def get_software_config(self, ud_content):
         try:
             sc = self.rpc_client().show_software_config(
@@ -1355,6 +1375,19 @@ class Server(stack_user.StackUser):
         except Exception as ex:
             self.client_plugin('swift').ignore_not_found(ex)
 
+    def _delete_queue(self):
+        queue_id = self.data().get('metadata_queue_id')
+        if not queue_id:
+            return
+        client_plugin = self.client_plugin('zaqar')
+        zaqar = client_plugin.create_for_tenant(
+            self.stack.stack_user_project_id)
+        try:
+            zaqar.queue(queue_id).delete()
+        except Exception as ex:
+            client_plugin.ignore_not_found(ex)
+        self.data_delete('metadata_queue_id')
+
     def handle_delete(self):
 
         if self.resource_id is None:
@@ -1363,6 +1396,7 @@ class Server(stack_user.StackUser):
         if self.user_data_software_config():
             self._delete_user()
             self._delete_temp_url()
+            self._delete_queue()
 
         try:
             server = self.nova().servers.get(self.resource_id)

heat/engine/service_software_config.py

@@ -71,7 +71,7 @@ class SoftwareConfigService(service.Service):
         result = [api.format_software_config(sd.config) for sd in all_sd_s]
         return result
 
-    def _push_metadata_software_deployments(self, cnxt, server_id):
+    def _push_metadata_software_deployments(self, cnxt, server_id, sd):
         rs = (resource_object.Resource.
               get_by_physical_resource_id(cnxt, server_id))
         if not rs:
@@ -82,15 +82,24 @@ class SoftwareConfigService(service.Service):
         rs.update_and_save({'rsrc_metadata': md})
 
         metadata_put_url = None
+        metadata_queue_id = None
         for rd in rs.data:
             if rd.key == 'metadata_put_url':
                 metadata_put_url = rd.value
                 break
+            elif rd.key == 'metadata_queue_id':
+                metadata_queue_id = rd.value
+                break
         if metadata_put_url:
             json_md = jsonutils.dumps(md)
             requests.put(metadata_put_url, json_md)
+        elif metadata_queue_id:
+            zaqar_plugin = cnxt.clients.client_plugin('zaqar')
+            zaqar = zaqar_plugin.create_for_tenant(sd.stack_user_project_id)
+            queue = zaqar.queue(metadata_queue_id)
+            queue.post({'body': md, 'ttl': zaqar_plugin.DEFAULT_TTL})
 
-    def _refresh_software_deployment(self, cnxt, sd, deploy_signal_id):
+    def _refresh_swift_software_deployment(self, cnxt, sd, deploy_signal_id):
         container, object_name = urlparse.urlparse(
             deploy_signal_id).path.split('/')[-2:]
         swift_plugin = cnxt.clients.client_plugin('swift')
@@ -136,6 +145,19 @@ class SoftwareConfigService(service.Service):
         return software_deployment_object.SoftwareDeployment.get_by_id(
             cnxt, sd.id)
 
+    def _refresh_zaqar_software_deployment(self, cnxt, sd, deploy_queue_id):
+        zaqar_plugin = cnxt.clients.client_plugin('zaqar')
+        zaqar = zaqar_plugin.create_for_tenant(sd.stack_user_project_id)
+        queue = zaqar.queue(deploy_queue_id)
+
+        messages = list(queue.pop())
+        if messages:
+            self.signal_software_deployment(
+                cnxt, sd.id, messages[0].body, None)
+
+        return software_deployment_object.SoftwareDeployment.get_by_id(
+            cnxt, sd.id)
+
     def show_software_deployment(self, cnxt, deployment_id):
         sd = software_deployment_object.SoftwareDeployment.get_by_id(
             cnxt, deployment_id)
@@ -144,8 +166,11 @@ class SoftwareConfigService(service.Service):
             input_values = dict((i['name'], i['value']) for i in c['inputs'])
             transport = input_values.get('deploy_signal_transport')
             if transport == 'TEMP_URL_SIGNAL':
-                sd = self._refresh_software_deployment(
+                sd = self._refresh_swift_software_deployment(
                     cnxt, sd, input_values.get('deploy_signal_id'))
+            elif transport == 'ZAQAR_SIGNAL':
+                sd = self._refresh_zaqar_software_deployment(
+                    cnxt, sd, input_values.get('deploy_queue_id'))
         return api.format_software_deployment(sd)
 
     def create_software_deployment(self, cnxt, server_id, config_id,
@@ -161,7 +186,7 @@ class SoftwareConfigService(service.Service):
             'action': action,
             'status': status,
             'status_reason': status_reason})
-        self._push_metadata_software_deployments(cnxt, server_id)
+        self._push_metadata_software_deployments(cnxt, server_id, sd)
         return api.format_software_deployment(sd)
 
     def signal_software_deployment(self, cnxt, deployment_id, details,
@@ -252,7 +277,7 @@ class SoftwareConfigService(service.Service):
         # only push metadata if this update resulted in the config_id
         # changing, since metadata is just a list of configs
         if config_id:
-            self._push_metadata_software_deployments(cnxt, sd.server_id)
+            self._push_metadata_software_deployments(cnxt, sd.server_id, sd)
 
         return api.format_software_deployment(sd)
 

heat/tests/engine/test_software_config.py

@@ -23,6 +23,7 @@ import six
 from heat.common import exception
 from heat.common import template_format
 from heat.engine.clients.os import swift
+from heat.engine.clients.os import zaqar
 from heat.engine import service
 from heat.engine import service_software_config
 from heat.objects import resource as resource_objects
@@ -376,9 +377,9 @@ class SoftwareConfigServiceTest(common.HeatTestCase):
         self.assertEqual(kwargs['input_values'], deployment['input_values'])
 
     @mock.patch.object(service_software_config.SoftwareConfigService,
-                       '_refresh_software_deployment')
+                       '_refresh_swift_software_deployment')
     def test_show_software_deployment_refresh(
-            self, _refresh_software_deployment):
+            self, _refresh_swift_software_deployment):
         temp_url = ('http://192.0.2.1/v1/AUTH_a/b/c'
                     '?temp_url_sig=ctemp_url_expires=1234')
         config = self._create_software_config(inputs=[
@@ -399,13 +400,13 @@ class SoftwareConfigServiceTest(common.HeatTestCase):
         deployment_id = deployment['id']
         sd = software_deployment_object.SoftwareDeployment.get_by_id(
             self.ctx, deployment_id)
-        _refresh_software_deployment.return_value = sd
+        _refresh_swift_software_deployment.return_value = sd
         self.assertEqual(
             deployment,
             self.engine.show_software_deployment(self.ctx, deployment_id))
         self.assertEqual(
             (self.ctx, sd, temp_url),
-            _refresh_software_deployment.call_args[0])
+            _refresh_swift_software_deployment.call_args[0])
 
     def test_update_software_deployment_new_config(self):
 
@@ -449,7 +450,10 @@ class SoftwareConfigServiceTest(common.HeatTestCase):
         self.assertIsNotNone(updated)
         self.assertEqual('DEPLOY', updated['action'])
         self.assertEqual('WAITING', updated['status'])
-        mock_push.assert_called_once_with(self.ctx, server_id)
+
+        sd = software_deployment_object.SoftwareDeployment.get_by_id(
+            self.ctx, deployment_id)
+        mock_push.assert_called_once_with(self.ctx, server_id, sd)
 
     def test_update_software_deployment_fields(self):
 
@@ -519,7 +523,7 @@ class SoftwareConfigServiceTest(common.HeatTestCase):
         }
 
         self.engine.software_config._push_metadata_software_deployments(
-            self.ctx, '1234')
+            self.ctx, '1234', None)
         rs.update_and_save.assert_called_once_with(
             {'rsrc_metadata': result_metadata})
         put.side_effect = Exception('Unexpected requests.put')
@@ -548,17 +552,56 @@ class SoftwareConfigServiceTest(common.HeatTestCase):
         }
 
         self.engine.software_config._push_metadata_software_deployments(
-            self.ctx, '1234')
+            self.ctx, '1234', None)
         rs.update_and_save.assert_called_once_with(
             {'rsrc_metadata': result_metadata})
 
         put.assert_called_once_with(
             'http://192.168.2.2/foo/bar', json.dumps(result_metadata))
 
+    @mock.patch.object(service_software_config.SoftwareConfigService,
+                       'metadata_software_deployments')
+    @mock.patch.object(service_software_config.resource_object.Resource,
+                       'get_by_physical_resource_id')
+    @mock.patch.object(zaqar.ZaqarClientPlugin, 'create_for_tenant')
+    def test_push_metadata_software_deployments_queue(
+            self, plugin, res_get, md_sd):
+        rs = mock.Mock()
+        rs.rsrc_metadata = {'original': 'metadata'}
+        rd = mock.Mock()
+        rd.key = 'metadata_queue_id'
+        rd.value = '6789'
+        rs.data = [rd]
+        res_get.return_value = rs
+        sd = mock.Mock()
+        sd.stack_user_project_id = 'project1'
+        queue = mock.Mock()
+        zaqar_client = mock.Mock()
+        plugin.return_value = zaqar_client
+        zaqar_client.queue.return_value = queue
+
+        deployments = {'deploy': 'this'}
+        md_sd.return_value = deployments
+
+        result_metadata = {
+            'original': 'metadata',
+            'deployments': {'deploy': 'this'}
+        }
+
+        self.engine.software_config._push_metadata_software_deployments(
+            self.ctx, '1234', sd)
+        rs.update_and_save.assert_called_once_with(
+            {'rsrc_metadata': result_metadata})
+
+        plugin.assert_called_once_with('project1')
+        zaqar_client.queue.assert_called_once_with('6789')
+        queue.post.assert_called_once_with(
+            {'body': result_metadata, 'ttl': 3600})
+
     @mock.patch.object(service_software_config.SoftwareConfigService,
                        'signal_software_deployment')
     @mock.patch.object(swift.SwiftClientPlugin, '_create')
-    def test_refresh_software_deployment(self, scc, ssd):
+    def test_refresh_swift_software_deployment(self, scc, ssd):
         temp_url = ('http://192.0.2.1/v1/AUTH_a/b/c'
                     '?temp_url_sig=ctemp_url_expires=1234')
         container = 'b'
@@ -607,7 +650,7 @@ class SoftwareConfigServiceTest(common.HeatTestCase):
 
         self.assertEqual(
             sd,
-            self.engine.software_config._refresh_software_deployment(
+            self.engine.software_config._refresh_swift_software_deployment(
                 self.ctx, sd, temp_url))
         sc.head_object.assert_called_once_with(container, object_name)
         # no call to get_object or signal_last_modified
@@ -619,7 +662,7 @@ class SoftwareConfigServiceTest(common.HeatTestCase):
             'Ouch', http_status=409)
         self.assertRaises(
             swift_exc.ClientException,
-            self.engine.software_config._refresh_software_deployment,
+            self.engine.software_config._refresh_swift_software_deployment,
             self.ctx,
             sd,
             temp_url)
@@ -629,7 +672,7 @@ class SoftwareConfigServiceTest(common.HeatTestCase):
         sc.head_object.side_effect = None
 
         # first poll populates data signal_last_modified
-        self.engine.software_config._refresh_software_deployment(
+        self.engine.software_config._refresh_swift_software_deployment(
             self.ctx, sd, temp_url)
         sc.head_object.assert_called_with(container, object_name)
         sc.get_object.assert_called_once_with(container, object_name)
@@ -643,7 +686,7 @@ class SoftwareConfigServiceTest(common.HeatTestCase):
         sd = software_deployment_object.SoftwareDeployment.get_by_id(
             self.ctx, deployment_id)
         self.assertEqual(then, sd.updated_at)
-        self.engine.software_config._refresh_software_deployment(
+        self.engine.software_config._refresh_swift_software_deployment(
             self.ctx, sd, temp_url)
         sc.get_object.assert_called_once_with(container, object_name)
         # signal_software_deployment has not been called again
@@ -654,7 +697,7 @@ class SoftwareConfigServiceTest(common.HeatTestCase):
         headers['last-modified'] = last_modified_2
         sc.head_object.return_value = headers
         sc.get_object.return_value = (headers, '{"bar": "baz"}')
-        self.engine.software_config._refresh_software_deployment(
+        self.engine.software_config._refresh_swift_software_deployment(
             self.ctx, sd, temp_url)
 
         # two calls to signal_software_deployment, for then and now
@@ -667,6 +710,40 @@ class SoftwareConfigServiceTest(common.HeatTestCase):
             self.ctx, deployment_id, {'updated_at': now})
         sd = software_deployment_object.SoftwareDeployment.get_by_id(
             self.ctx, deployment_id)
-        self.engine.software_config._refresh_software_deployment(
+        self.engine.software_config._refresh_swift_software_deployment(
             self.ctx, sd, temp_url)
         self.assertEqual(2, len(ssd.mock_calls))
+
+    @mock.patch.object(service_software_config.SoftwareConfigService,
+                       'signal_software_deployment')
+    @mock.patch.object(zaqar.ZaqarClientPlugin, 'create_for_tenant')
+    def test_refresh_zaqar_software_deployment(self, plugin, ssd):
+        config = self._create_software_config(inputs=[
+            {
+                'name': 'deploy_signal_transport',
+                'type': 'String',
+                'value': 'ZAQAR_SIGNAL'
+            }, {
+                'name': 'deploy_queue_id',
+                'type': 'String',
+                'value': '6789'
+            }
+        ])
+
+        queue = mock.Mock()
+        zaqar_client = mock.Mock()
+        plugin.return_value = zaqar_client
+        zaqar_client.queue.return_value = queue
+        queue.pop.return_value = [mock.Mock(body='ok')]
+
+        deployment = self._create_software_deployment(
+            status='IN_PROGRESS', config_id=config['id'])
+
+        deployment_id = deployment['id']
+        self.assertEqual(
+            deployment,
+            self.engine.show_software_deployment(self.ctx, deployment_id))
+
+        zaqar_client.queue.assert_called_once_with('6789')
+        queue.pop.assert_called_once()
+        ssd.assert_called_once_with(self.ctx, deployment_id, 'ok', None)

heat/tests/test_server.py

@@ -27,6 +27,7 @@ from heat.engine.clients.os import glance
 from heat.engine.clients.os import neutron
 from heat.engine.clients.os import nova
 from heat.engine.clients.os import swift
+from heat.engine.clients.os import zaqar
 from heat.engine import environment
 from heat.engine import resource
 from heat.engine.resources.openstack.nova import server as servers
@@ -820,6 +821,75 @@ class ServersTest(common.HeatTestCase):
 
         self.m.VerifyAll()
 
+    def test_server_create_software_config_zaqar(self):
+        return_server = self.fc.servers.list()[1]
+        stack_name = 'software_config_s'
+        (tmpl, stack) = self._setup_test_stack(stack_name)
+
+        props = tmpl.t['Resources']['WebServer']['Properties']
+        props['user_data_format'] = 'SOFTWARE_CONFIG'
+        props['software_config_transport'] = 'ZAQAR_MESSAGE'
+
+        resource_defns = tmpl.resource_definitions(stack)
+        server = servers.Server('WebServer',
+                                resource_defns['WebServer'], stack)
+
+        ncp = self.patchobject(nova.NovaClientPlugin, '_create')
+        zcc = self.patchobject(zaqar.ZaqarClientPlugin, 'create_for_tenant')
+        zc = mock.Mock()
+
+        ncp.return_value = self.fc
+        zcc.return_value = zc
+        queue = mock.Mock()
+        zc.queue.return_value = queue
+        self._mock_get_image_id_success('F17-x86_64-gold', 744)
+
+        self.m.StubOutWithMock(self.fc.servers, 'create')
+        self.fc.servers.create(
+            image=744, flavor=3, key_name='test',
+            name=utils.PhysName(stack_name, server.name),
+            security_groups=[],
+            userdata=mox.IgnoreArg(), scheduler_hints=None,
+            meta=None, nics=None, availability_zone=None,
+            block_device_mapping=None, block_device_mapping_v2=None,
+            config_drive=None, disk_config=None, reservation_id=None,
+            files={}, admin_pass=None).AndReturn(
+                return_server)
+
+        self.m.ReplayAll()
+        scheduler.TaskRunner(server.create)()
+
+        metadata_queue_id = server.data().get('metadata_queue_id')
+        md = server.metadata_get()
+        queue_id = md['os-collect-config']['zaqar']['queue_id']
+        self.assertEqual(queue_id, metadata_queue_id)
+
+        self.assertEqual({
+            'os-collect-config': {
+                'zaqar': {
+                    'user_id': '1234',
+                    'password': server.password,
+                    'auth_url': 'http://server.test:5000/v2.0',
+                    'project_id': '8888',
+                    'queue_id': queue_id
+                }
+            },
+            'deployments': []
+        }, server.metadata_get())
+
+        zc.queue.assert_called_once_with(queue_id)
+        queue.post.assert_called_once_with(
+            {'body': server.metadata_get(), 'ttl': 3600})
+
+        zc.queue.reset_mock()
+
+        server._delete_queue()
+
+        zc.queue.assert_called_once_with(queue_id)
+        zc.queue.delete.assert_called_once()
+
+        self.m.VerifyAll()
+
     @mock.patch.object(nova.NovaClientPlugin, '_create')
     def test_server_create_default_admin_pass(self, mock_client):
         return_server = self.fc.servers.list()[1]

heat/tests/test_software_deployment.py

@@ -22,6 +22,7 @@ from heat.common import exception as exc
 from heat.common.i18n import _
 from heat.engine.clients.os import nova
 from heat.engine.clients.os import swift
+from heat.engine.clients.os import zaqar
 from heat.engine.resources.openstack.heat import software_deployment as sd
 from heat.engine import rsrc_defn
 from heat.engine import stack as parser
@@ -100,6 +101,22 @@ class SoftwareDeploymentTest(common.HeatTestCase):
         }
     }
 
+    template_zaqar_signal = {
+        'HeatTemplateFormatVersion': '2012-12-12',
+        'Resources': {
+            'deployment_mysql': {
+                'Type': 'OS::Heat::SoftwareDeployment',
+                'Properties': {
+                    'server': '9f1f0e00-05d2-4ca5-8602-95021f19c9d0',
+                    'config': '48e8ade1-9196-42d5-89a2-f709fde42632',
+                    'input_values': {'foo': 'bar', 'bink': 'bonk'},
+                    'signal_transport': 'ZAQAR_SIGNAL',
+                    'name': '00_run_me_first'
+                }
+            }
+        }
+    }
+
     template_delete_suspend_resume = {
         'HeatTemplateFormatVersion': '2012-12-12',
         'Resources': {
@@ -1025,6 +1042,69 @@ class SoftwareDeploymentTest(common.HeatTestCase):
         for action in ('CREATE', 'UPDATE'):
             self.assertIsNotNone(self.deployment._handle_action(action))
 
+    def test_get_zaqar_queue(self):
+        dep_data = {}
+
+        zc = mock.MagicMock()
+        zcc = self.patch(
+            'heat.engine.clients.os.zaqar.ZaqarClientPlugin._create')
+        zcc.return_value = zc
+
+        self._create_stack(self.template_zaqar_signal)
+
+        def data_set(key, value, redact=False):
+            dep_data[key] = value
+
+        self.deployment.data_set = data_set
+        self.deployment.data = mock.Mock(return_value=dep_data)
+
+        self.deployment.id = 23
+        self.deployment.uuid = str(uuid.uuid4())
+        self.deployment.action = self.deployment.CREATE
+
+        queue_id = self.deployment._get_queue_id()
+        self.assertEqual(2, len(zc.queue.mock_calls))
+        self.assertEqual(queue_id, zc.queue.mock_calls[0][1][0])
+        self.assertEqual(queue_id, dep_data['signal_queue_id'])
+
+        self.assertEqual(queue_id, self.deployment._get_queue_id())
+
+    def test_delete_zaqar_queue(self):
+        queue_id = str(uuid.uuid4())
+        dep_data = {
+            'signal_queue_id': queue_id
+        }
+        self._create_stack(self.template_zaqar_signal)
+
+        self.deployment.data_delete = mock.MagicMock()
+        self.deployment.data = mock.Mock(return_value=dep_data)
+
+        zc = mock.MagicMock()
+        zcc = self.patch(
+            'heat.engine.clients.os.zaqar.ZaqarClientPlugin._create')
+        zcc.return_value = zc
+
+        self.deployment.id = 23
+        self.deployment.uuid = str(uuid.uuid4())
+        self.deployment._delete_queue()
+        zc.queue.assert_called_once_with(queue_id)
+        zc.queue.delete.assert_called_once()
+        self.assertEqual(
+            [mock.call('signal_queue_id')],
+            self.deployment.data_delete.mock_calls)
+
+        zaqar_exc = zaqar.ZaqarClientPlugin.exceptions_module
+        zc.queue.delete.side_effect = zaqar_exc.ResourceNotFound()
+        self.deployment._delete_queue()
+        self.assertEqual(
+            [mock.call('signal_queue_id'), mock.call('signal_queue_id')],
+            self.deployment.data_delete.mock_calls)
+
+        dep_data.pop('signal_queue_id')
+        self.deployment.physical_resource_name = mock.Mock()
+        self.deployment._delete_queue()
+        self.assertEqual(2, len(self.deployment.data_delete.mock_calls))
+
 
 class SoftwareDeploymentGroupTest(common.HeatTestCase):