6b07d55181
Zaqar queues were accessed and created with the user provided token, but we need them to be in the stack project for instances to access them. This changes to provide a stack user token instead. Change-Id: If30ad4ad63b845b8dabe0b3ac0e4209057c3753c Closes-Bug: #1532235
51 lines
1.9 KiB
Python
51 lines
1.9 KiB
Python
#
|
|
# 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 mock
|
|
|
|
from heat.engine.clients.os import zaqar
|
|
|
|
from heat.tests import common
|
|
from heat.tests import utils
|
|
|
|
|
|
class ZaqarClientPluginTests(common.HeatTestCase):
|
|
|
|
def test_create(self):
|
|
context = utils.dummy_context()
|
|
plugin = context.clients.client_plugin('zaqar')
|
|
client = plugin.client()
|
|
self.assertEqual('http://server.test:5000/v3', client.api_url)
|
|
self.assertEqual(1.1, client.api_version)
|
|
self.assertEqual('test_tenant_id',
|
|
client.conf['auth_opts']['options']['os_project_id'])
|
|
|
|
def test_create_for_tenant(self):
|
|
context = utils.dummy_context()
|
|
plugin = context.clients.client_plugin('zaqar')
|
|
client = plugin.create_for_tenant('other_tenant', 'token')
|
|
self.assertEqual('other_tenant',
|
|
client.conf['auth_opts']['options']['os_project_id'])
|
|
self.assertEqual('token',
|
|
client.conf['auth_opts']['options']['os_auth_token'])
|
|
|
|
def test_event_sink(self):
|
|
context = utils.dummy_context()
|
|
client = context.clients.client('zaqar')
|
|
fake_queue = mock.MagicMock()
|
|
client.queue = lambda x, auto_create: fake_queue
|
|
sink = zaqar.ZaqarEventSink('myqueue')
|
|
sink.consume(context, {'hello': 'world'})
|
|
fake_queue.post.assert_called_once_with(
|
|
{'body': {'hello': 'world'}, 'ttl': 3600})
|