openstack/python-zaqarclient
Code Issues Proposed changes
Files
47f1bfcb61a50c0ce25e8dc0a48c3ec983e769b7
python-zaqarclient/zaqarclient/tests/queues/queues.py
Fei Long Wang 47f1bfcb61 Add subscriptions for queue object 
A new method 'subscriptions' is added for queue object in this
patch, so that the API user can list subscriptions of a queue
easily.

Change-Id: Ia4a3da6a1b2165ba623c7101ecb4614245520f1b
2016-02-18 12:07:14 +13:00

538 lines
18 KiB
Python
Raw Blame History

# Copyright (c) 2013 Red Hat, Inc.
#
# 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 json
import mock
from zaqarclient import errors
from zaqarclient.queues import client
from zaqarclient.queues.v1 import iterator
from zaqarclient.queues.v1 import message
from zaqarclient.queues.v2 import subscription
from zaqarclient.tests.queues import base
from zaqarclient.transport import response
class QueuesV1QueueUnitTest(base.QueuesTestBase):
def test_queue_metadata(self):
test_metadata = {'type': 'Bank Accounts'}
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, json.dumps(test_metadata))
send_method.return_value = resp
metadata = self.queue.metadata(test_metadata)
self.assertEqual(test_metadata, metadata)
def test_queue_metadata_update(self):
test_metadata = {'type': 'Bank Accounts'}
new_meta = {'flavor': 'test'}
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, json.dumps(test_metadata))
send_method.return_value = resp
metadata = self.queue.metadata(test_metadata)
self.assertEqual(test_metadata, metadata)
metadata = self.queue.metadata(new_meta)
self.assertEqual(new_meta, metadata)
def test_queue_create(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, None)
send_method.return_value = resp
self.queue.ensure_exists()
# NOTE(flaper87): Nothing to assert here,
# just checking our way down to the transport
# doesn't crash.
def test_queue_valid_name(self):
self.assertRaises(ValueError, self.client.queue, "")
def test_queue_delete(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, None)
send_method.return_value = resp
self.queue.delete()
# NOTE(flaper87): Nothing to assert here,
# just checking our way down to the transport
# doesn't crash.
def test_queue_exists(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, None)
send_method.return_value = resp
self.queue.exists()
# NOTE(flaper87): Nothing to assert here,
# just checking our way down to the transport
# doesn't crash.
def test_queue_stats(self):
result = {
"messages": {
"free": 146929,
"claimed": 2409,
"total": 149338,
"oldest": {
"href": "/v1/queues/qq/messages/50b68a50d6f5b8c8a7c62b01",
"age": 63,
"created": "2013-08-12T20:44:55Z"
},
"newest": {
"href": "/v1/queues/qq/messages/50b68a50d6f5b8c8a7c62b01",
"age": 12,
"created": "2013-08-12T20:45:46Z"
}
}
}
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, json.dumps(result))
send_method.return_value = resp
stats = self.queue.stats
self.assertEqual(result, stats)
def test_message_post(self):
messages = [{'ttl': 30, 'body': 'Post It!'}]
result = {
"resources": [
"/v1/queues/fizbit/messages/50b68a50d6f5b8c8a7c62b01"
],
"partial": False
}
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, json.dumps(result))
send_method.return_value = resp
posted = self.queue.post(messages)
self.assertEqual(result, posted)
def test_message_list(self):
returned = {
'links': [{
'rel': 'next',
'href': '/v1/queues/fizbit/messages?marker=6244-244224-783'
}],
'messages': [{
'href': '/v1/queues/fizbit/messages/50b68a50d6f5b8c8a7c62b01',
'ttl': 800,
'age': 790,
'body': {'event': 'ActivateAccount',
'mode': 'active'}
}]
}
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, json.dumps(returned))
send_method.return_value = resp
msgs = self.queue.messages(limit=1)
self.assertIsInstance(msgs, iterator._Iterator)
# NOTE(flaper87): Nothing to assert here,
# just checking our way down to the transport
# doesn't crash.
def test_message_get(self):
returned = {
'href': '/v1/queues/fizbit/messages/50b68a50d6f5b8c8a7c62b01',
'ttl': 800,
'age': 790,
'body': {'event': 'ActivateAccount', 'mode': 'active'}
}
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, json.dumps(returned))
send_method.return_value = resp
msgs = self.queue.message('50b68a50d6f5b8c8a7c62b01')
self.assertIsInstance(msgs, message.Message)
# NOTE(flaper87): Nothing to assert here,
# just checking our way down to the transport
# doesn't crash.
def test_message_get_many(self):
returned = [{
'href': '/v1/queues/fizbit/messages/50b68a50d6f5b8c8a7c62b01',
'ttl': 800,
'age': 790,
'body': {'event': 'ActivateAccount', 'mode': 'active'}
}, {
'href': '/v1/queues/fizbit/messages/50b68a50d6f5b8c8a7c62b02',
'ttl': 800,
'age': 790,
'body': {'event': 'ActivateAccount', 'mode': 'active'}
}]
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, json.dumps(returned))
send_method.return_value = resp
msg = self.queue.messages('50b68a50d6f5b8c8a7c62b01',
'50b68a50d6f5b8c8a7c62b02')
self.assertIsInstance(msg, iterator._Iterator)
# NOTE(flaper87): Nothing to assert here,
# just checking our way down to the transport
# doesn't crash.
def test_message_delete_many(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, None)
send_method.return_value = resp
rst = self.queue.delete_messages('50b68a50d6f5b8c8a7c62b01',
'50b68a50d6f5b8c8a7c62b02')
self.assertIsNone(rst)
# NOTE(flaper87): Nothing to assert here,
# just checking our way down to the transport
# doesn't crash.
class QueuesV1QueueFunctionalTest(base.QueuesTestBase):
def test_queue_create_functional(self):
queue = self.client.queue("nonono")
self.addCleanup(queue.delete)
queue._get_transport = mock.Mock(return_value=self.transport)
self.assertTrue(queue.exists())
def test_queue_delete_functional(self):
queue = self.client.queue("nonono")
self.addCleanup(queue.delete)
queue._get_transport = mock.Mock(return_value=self.transport)
self.assertTrue(queue.exists())
queue.delete()
self.assertFalse(queue.exists())
def test_queue_exists_functional(self):
queue = self.client.queue("404", auto_create=False)
queue._get_transport = mock.Mock(return_value=self.transport)
self.assertFalse(queue.exists())
def test_queue_stats_functional(self):
messages = [
{'ttl': 60, 'body': 'Post It!'},
{'ttl': 60, 'body': 'Post It!'},
{'ttl': 60, 'body': 'Post It!'},
]
queue = self.client.queue("nonono")
self.addCleanup(queue.delete)
queue._get_transport = mock.Mock(return_value=self.transport)
queue.post(messages)
stats = queue.stats
self.assertEqual(3, stats["messages"]["free"])
def test_queue_metadata_functional(self):
test_metadata = {'type': 'Bank Accounts'}
queue = self.client.queue("meta-test")
queue.metadata(test_metadata)
# NOTE(flaper87): Clear metadata's cache
queue._metadata = None
metadata = queue.metadata()
self.assertEqual(test_metadata, metadata)
def test_queue_metadata_reload_functional(self):
test_metadata = {'type': 'Bank Accounts'}
queue = self.client.queue("meta-test")
queue.metadata(test_metadata)
# NOTE(flaper87): Overwrite the cached value
# but don't clear it.
queue._metadata = 'test'
metadata = queue.metadata(force_reload=True)
self.assertEqual(test_metadata, metadata)
def test_message_post_functional(self):
messages = [
{'ttl': 60, 'body': 'Post It!'},
{'ttl': 60, 'body': 'Post It!'},
{'ttl': 60, 'body': 'Post It!'},
]
queue = self.client.queue("nonono")
queue._get_transport = mock.Mock(return_value=self.transport)
result = queue.post(messages)
self.assertIn('resources', result)
self.assertEqual(3, len(result['resources']))
def test_message_list_functional(self):
queue = self.client.queue("test_queue")
self.addCleanup(queue.delete)
queue._get_transport = mock.Mock(return_value=self.transport)
messages = [{'ttl': 60, 'body': 'Post It 1!'}]
queue.post(messages)
messages = queue.messages()
self.assertTrue(isinstance(messages, iterator._Iterator))
self.assertGreaterEqual(len(list(messages)), 0)
def test_message_list_echo_functional(self):
queue = self.client.queue("test_queue")
self.addCleanup(queue.delete)
queue._get_transport = mock.Mock(return_value=self.transport)
messages = [
{'ttl': 60, 'body': 'Post It 1!'},
{'ttl': 60, 'body': 'Post It 2!'},
{'ttl': 60, 'body': 'Post It 3!'},
]
queue.post(messages)
messages = queue.messages(echo=True)
self.assertTrue(isinstance(messages, iterator._Iterator))
self.assertGreaterEqual(len(list(messages)), 3)
def test_message_get_functional(self):
queue = self.client.queue("test_queue")
self.addCleanup(queue.delete)
queue._get_transport = mock.Mock(return_value=self.transport)
messages = [
{'ttl': 60, 'body': 'Post It 1!'},
{'ttl': 60, 'body': 'Post It 2!'},
{'ttl': 60, 'body': 'Post It 3!'},
]
res = queue.post(messages)['resources']
msg_id = res[0].split('/')[-1]
msg = queue.message(msg_id)
self.assertTrue(isinstance(msg, message.Message))
self.assertEqual(res[0], msg.href)
def test_message_get_many_functional(self):
queue = self.client.queue("test_queue")
self.addCleanup(queue.delete)
queue._get_transport = mock.Mock(return_value=self.transport)
messages = [
{'ttl': 60, 'body': 'Post It 1!'},
{'ttl': 60, 'body': 'Post It 2!'},
{'ttl': 60, 'body': 'Post It 3!'},
]
res = queue.post(messages)['resources']
msgs_id = [ref.split('/')[-1] for ref in res]
messages = queue.messages(*msgs_id)
self.assertTrue(isinstance(messages, iterator._Iterator))
messages = list(messages)
length = len(messages)
if length == 3:
bodies = set(message.body for message in messages)
self.assertEqual(
set(['Post It 1!', 'Post It 2!', 'Post It 3!']), bodies)
elif length == 1:
# FIXME(therve): Old broken behavior, remove it as some point
pass
else:
self.fail("Wrong number of messages: '%d'" % length)
def test_message_delete_many_functional(self):
queue = self.client.queue("test_queue")
self.addCleanup(queue.delete)
queue._get_transport = mock.Mock(return_value=self.transport)
messages = [
{'ttl': 60, 'body': 'Post It 1!'},
{'ttl': 60, 'body': 'Post It 2!'},
]
res = queue.post(messages)['resources']
msgs_id = [ref.split('/')[-1] for ref in res]
queue.delete_messages(*msgs_id)
messages = queue.messages()
self.assertEqual(0, len(list(messages)))
class QueuesV1_1QueueUnitTest(QueuesV1QueueUnitTest):
def test_message_pop(self):
returned = [{
'href': '/v1/queues/fizbit/messages/50b68a50d6f5b8c8a7c62b01',
'ttl': 800,
'age': 790,
'body': {'event': 'ActivateAccount', 'mode': 'active'}
}, {
'href': '/v1/queues/fizbit/messages/50b68a50d6f5b8c8a7c62b02',
'ttl': 800,
'age': 790,
'body': {'event': 'ActivateAccount', 'mode': 'active'}
}]
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, json.dumps(returned))
send_method.return_value = resp
msg = self.queue.pop(count=2)
self.assertIsInstance(msg, iterator._Iterator)
# NOTE(flaper87): Nothing to assert here,
# just checking our way down to the transport
# doesn't crash.
class QueuesV1_1QueueFunctionalTest(QueuesV1QueueFunctionalTest):
def test_queue_create_functional(self):
pass
def test_queue_exists_functional(self):
queue = self.client.queue("404")
self.assertRaises(errors.InvalidOperation, queue.exists)
def test_queue_delete_functional(self):
queue = self.client.queue("nonono")
queue._get_transport = mock.Mock(return_value=self.transport)
messages = [
{'ttl': 60, 'body': 'Post It 1!'},
{'ttl': 60, 'body': 'Post It 2!'},
{'ttl': 60, 'body': 'Post It 3!'},
]
queue.post(messages)
queue.delete()
self.assertEqual(0, len(list(queue.messages(echo=True))))
def test_message_pop(self):
queue = self.client.queue("test_queue")
self.addCleanup(queue.delete)
queue._get_transport = mock.Mock(return_value=self.transport)
messages = [
{'ttl': 60, 'body': 'Post It 1!'},
{'ttl': 60, 'body': 'Post It 2!'},
{'ttl': 60, 'body': 'Post It 2!'},
]
queue.post(messages)
messages = queue.pop(count=2)
self.assertTrue(isinstance(messages, iterator._Iterator))
self.assertEqual(2, len(list(messages)))
remaining = queue.messages(echo=True)
self.assertEqual(1, len(list(remaining)))
class QueuesV2QueueUnitTest(QueuesV1_1QueueUnitTest):
def test_message_get(self):
pass
def test_queue_subscriptions(self):
result = {
"subscriptions": [{
"source": 'test',
"id": "1",
"subscriber": 'http://trigger.me',
"ttl": 3600,
"options": {}},
{
"source": 'test',
"id": "2",
"subscriber": 'http://trigger.you',
"ttl": 7200,
"options": {}}]
}
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, json.dumps(result))
send_method.return_value = resp
subscriptions = self.queue.subscriptions()
subscriber_list = [s.subscriber for s in list(subscriptions)]
self.assertIn('http://trigger.me', subscriber_list)
self.assertIn('http://trigger.you', subscriber_list)
class QueuesV2QueueFunctionalTest(QueuesV1_1QueueFunctionalTest):
def test_signed_url(self):
queue = self.client.queue('test_queue')
messages = [{'ttl': 300, 'body': 'Post It!'}]
queue.post(messages)
self.addCleanup(queue.delete)
signature = queue.signed_url()
opts = {
'paths': signature['paths'],
'expires': signature['expires'],
'methods': signature['methods'],
'signature': signature['signature'],
'os_project_id': signature['project'],
}
auth_opts = {'backend': 'signed-url',
'options': opts}
conf = {'auth_opts': auth_opts}
signed_client = client.Client(self.url, self.version, conf)
queue = signed_client.queue('test_queue')
[message] = list(queue.messages())
self.assertEqual('Post It!', message.body)
def test_queue_subscriptions(self):
queue_name = 'test_queue'
queue = self.client.queue(queue_name, force_create=True)
self.addCleanup(queue.delete)
queue._get_transport = mock.Mock(return_value=self.transport)
subscription.Subscription(self.client, queue_name,
subscriber='http://trigger.me')
subscription.Subscription(self.client, queue_name,
subscriber='http://trigger.you')
get_subscriptions = queue.subscriptions()
self.assertTrue(isinstance(get_subscriptions, iterator._Iterator))
self.assertEqual(2, len(list(get_subscriptions)))
View Git Blame Copy Permalink