openstack/python-zaqarclient
Code Issues Proposed changes

Remove v1 API support

Browse Source 
v1 API and v1.1 API were disabled some time ago and were both removed
recently. So we no longer have to carry support for these.

Change-Id: I75e2525d68af0f55cf629a1dc059bb162a6d483e
This commit is contained in:
Takashi Kajinami
2025-03-17 02:37:25 +09:00
parent 612e3dec9f
commit 887b2d7f73
45 changed files with 1808 additions and 2970 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
doc/source/client.rst
View File

@@ -12,11 +12,6 @@ Client Object Reference
This is the reference documentation for all API version.
API v1 and v1.1:
.. autoclass:: zaqarclient.queues.v1.client.Client
:members:
API v2.0:
.. autoclass:: zaqarclient.queues.v2.client.Client

4
releasenotes/notes/remove-v1-api-support-c0d2f4e130aa787a.yaml Normal file
View File

@@ -0,0 +1,4 @@
---
upgrade:
- |
Support for v1 API and v1.1 API has been removed.

10
setup.cfg
View File

@@ -30,22 +30,12 @@ packages =
[entry_points]
zaqarclient.transport =
http.v1 = zaqarclient.transport.http:HttpTransport
https.v1 = zaqarclient.transport.http:HttpTransport
http.v1.1 = zaqarclient.transport.http:HttpTransport
https.v1.1 = zaqarclient.transport.http:HttpTransport
http.v2 = zaqarclient.transport.http:HttpTransport
https.v2 = zaqarclient.transport.http:HttpTransport
ws.v1 = zaqarclient.transport.ws:WebsocketTransport
ws.v1.1 = zaqarclient.transport.ws:WebsocketTransport
ws.v2 = zaqarclient.transport.ws:WebsocketTransport
zaqarclient.api =
queues.v1 = zaqarclient.queues.v1.api:V1
queues.v1.1 = zaqarclient.queues.v1.api:V1_1
queues.v2 = zaqarclient.queues.v2.api:V2
openstack.messaging.v2 =

0
tests/functional/queues/v1/__init__.py
View File

32
tests/functional/queues/v1/test_claims.py
View File

@@ -1,32 +0,0 @@
# Copyright (c) 2014 Rackspace Hosting.
#
# 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.
from zaqarclient.tests.queues import claims
from zaqarclient.transport import http
class QueuesV1ClaimHttpFunctionalTest(claims.QueuesV1ClaimFunctionalTest):
is_functional = True
transport_cls = http.HttpTransport
version = 1
class QueuesV1_1ClaimHttpFunctionalTest(claims.QueuesV1_1ClaimFunctionalTest):
is_functional = True
transport_cls = http.HttpTransport
version = 1.1

25
tests/functional/queues/v1/test_flavor.py
View File

@@ -1,25 +0,0 @@
# Copyright (c) 2014 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.
from zaqarclient.tests.queues import flavor
from zaqarclient.transport import http
class QueuesV1_1FlavorHttpFunctionalTest(
flavor.QueuesV1_1FlavorFunctionalTest):
is_functional = True
transport_cls = http.HttpTransport
version = 1.1

25
tests/functional/queues/v1/test_pool.py
View File

@@ -1,25 +0,0 @@
# Copyright (c) 2014 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.
from zaqarclient.tests.queues import pool
from zaqarclient.transport import http
class QueuesV1_1PoolHttpFunctionalTest(pool.QueuesV1_1PoolFunctionalTest):
is_functional = True
transport_cls = http.HttpTransport
version = 1.1

32
tests/functional/queues/v1/test_queues.py
View File

@@ -1,32 +0,0 @@
# 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.
from zaqarclient.tests.queues import queues
from zaqarclient.transport import http
class QueuesV1QueueHttpFunctionalTest(queues.QueuesV1QueueFunctionalTest):
is_functional = True
transport_cls = http.HttpTransport
version = 1
class QueuesV1_1QueueHttpFunctionalTest(queues.QueuesV1_1QueueFunctionalTest):
is_functional = True
transport_cls = http.HttpTransport
version = 1.1

0
tests/unit/queues/v1/__init__.py
View File

32
tests/unit/queues/v1/test_claims.py
View File

@@ -1,32 +0,0 @@
# Copyright (c) Rackspace Hosting.
#
# 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.
from zaqarclient.tests.queues import claims
from zaqarclient.transport import http
class QueuesV1ClaimsHttpUnitTest(claims.QueueV1ClaimUnitTest):
transport_cls = http.HttpTransport
url = 'http://127.0.0.1:8888/v1'
version = 1
class QueuesV1_1ClaimsHttpUnitTest(claims.QueueV1_1ClaimUnitTest):
transport_cls = http.HttpTransport
url = 'http://127.0.0.1:8888/v1.1'
version = 1.1

55
tests/unit/queues/v1/test_client.py
View File

@@ -1,55 +0,0 @@
# Copyright 2014 IBM Corp.
#
# 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.
from unittest import mock
import ddt
from zaqarclient.queues import client
from zaqarclient.queues.v1 import core
from zaqarclient.tests import base
from zaqarclient.transport import errors
VERSIONS = [1, 1.1]
@ddt.ddt
class TestClient(base.TestBase):
@ddt.data(*VERSIONS)
def test_transport(self, version):
cli = client.Client('http://example.com',
version, {"auth_opts": {'backend': 'noauth'}})
self.assertIsNotNone(cli.transport())
@ddt.data(*VERSIONS)
def test_health_ok(self, version):
cli = client.Client('http://example.com',
version, {"auth_opts": {'backend': 'noauth'}})
with mock.patch.object(core, 'health', autospec=True) as core_health:
core_health.return_value = None
self.assertTrue(cli.health())
@ddt.data(*VERSIONS)
def test_health_bad(self, version):
cli = client.Client('http://example.com',
version, {"auth_opts": {'backend': 'noauth'}})
def raise_error(*args, **kwargs):
raise errors.ServiceUnavailableError()
with mock.patch.object(core, 'health', autospec=True) as core_health:
core_health.side_effect = raise_error
self.assertFalse(cli.health())

265
tests/unit/queues/v1/test_core.py
View File

@@ -1,265 +0,0 @@
# 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
from unittest import mock
from zaqarclient.queues.v1 import core
from zaqarclient.tests import base
from zaqarclient.tests.transport import dummy
from zaqarclient.transport import errors
from zaqarclient.transport import request
from zaqarclient.transport import response
class TestV1Core(base.TestBase):
def setUp(self):
super(TestV1Core, self).setUp()
self.transport = dummy.DummyTransport(self.conf)
def test_queue_create(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
send_method.return_value = response.Response(None, None)
req = request.Request()
core.queue_create(self.transport, req, 'test')
self.assertIn('queue_name', req.params)
def test_queue_delete(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
send_method.return_value = response.Response(None, None)
req = request.Request()
core.queue_delete(self.transport, req, 'test')
self.assertIn('queue_name', req.params)
def test_queue_exists(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
send_method.return_value = response.Response(None, None)
req = request.Request()
ret = core.queue_exists(self.transport, req, 'test')
self.assertIn('queue_name', req.params)
self.assertTrue(ret)
def test_queue_exists_not_found(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
send_method.side_effect = errors.ResourceNotFound
req = request.Request()
ret = core.queue_exists(self.transport, req, 'test')
self.assertIn('queue_name', req.params)
self.assertFalse(ret)
def test_get_queue_metadata(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, '{}')
send_method.return_value = resp
req = request.Request()
core.queue_get_metadata(self.transport, req, 'test')
def test_set_queue_metadata(self):
update_data = {'some': 'data'}
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
send_method.return_value = response.Response(None, None)
req = request.Request()
core.queue_exists(self.transport, req, update_data, 'test')
self.assertIn('queue_name', req.params)
def test_queue_get_stats(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, '{}')
send_method.return_value = resp
req = request.Request()
result = core.queue_get_stats(self.transport, req, 'test')
self.assertEqual({}, result)
def test_message_post_one(self):
messages = {'ttl': 30, 'body': 'Post one!'}
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, '{}')
send_method.return_value = resp
req = request.Request()
core.message_post(self.transport, req, 'test', messages)
self.assertIn('queue_name', req.params)
self.assertEqual(messages, json.loads(req.content))
def test_message_post_many(self):
messages = [{'ttl': 30, 'body': 'Post one!'},
{'ttl': 30, 'body': 'Post two!'},
{'ttl': 30, 'body': 'Post three!'}, ]
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, '{}')
send_method.return_value = resp
req = request.Request()
core.message_post(self.transport, req, 'test', messages)
self.assertIn('queue_name', req.params)
self.assertEqual(messages, json.loads(req.content))
def test_message_list(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, '{}')
send_method.return_value = resp
req = request.Request()
core.message_list(self.transport, req, 'test')
self.assertIn('queue_name', req.params)
def test_message_list_kwargs(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, '{}')
send_method.return_value = resp
req = request.Request()
core.message_list(self.transport, req, 'test',
marker='supermarket',
echo=False, limit=10)
self.assertIn('queue_name', req.params)
self.assertIn('limit', req.params)
self.assertIn('echo', req.params)
self.assertIn('marker', req.params)
def test_message_get_many(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, '{}')
send_method.return_value = resp
req = request.Request()
ids = ['a', 'b']
core.message_get_many(self.transport, req,
'test', ids)
self.assertIn('queue_name', req.params)
self.assertIn('ids', req.params)
self.assertEqual(ids, req.params['ids'])
def test_message_get(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, '{}')
send_method.return_value = resp
req = request.Request()
core.message_get(self.transport, req,
'test', 'message_id')
def test_message_delete(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, None)
send_method.return_value = resp
req = request.Request()
core.message_delete(self.transport, req,
'test', 'message_id')
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
ids = ['a', 'b']
req = request.Request()
core.message_delete_many(self.transport, req,
'test', ids=ids)
self.assertIn('queue_name', req.params)
self.assertIn('ids', req.params)
self.assertEqual(ids, req.params['ids'])
# ADMIN API
def test_pool_create(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, None)
send_method.return_value = resp
req = request.Request()
core.pool_create(self.transport, req,
'test_pool', {'uri': 'sqlite://',
'weight': 0})
def test_pool_get(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, None)
send_method.return_value = resp
req = request.Request()
core.pool_get(self.transport, req,
'test_pool')
def test_pool_delete(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, None)
send_method.return_value = resp
req = request.Request()
core.pool_delete(self.transport, req, 'test_pool')
def test_health(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, None)
send_method.return_value = resp
req = request.Request()
core.health(self.transport, req)
class TestV1_1Core(TestV1Core):
def test_message_pop(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, '{}')
send_method.return_value = resp
req = request.Request()
core.message_pop(self.transport, req,
'test', count=5)
self.assertIn('queue_name', req.params)
self.assertIn('pop', req.params)
self.assertEqual(5, req.params['pop'])

25
tests/unit/queues/v1/test_flavor.py
View File

@@ -1,25 +0,0 @@
# Copyright (c) 2014 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.
from zaqarclient.tests.queues import flavor
from zaqarclient.transport import http
class QueuesV1_1FlavorHttpUnitTest(flavor.QueuesV1_1FlavorUnitTest):
transport_cls = http.HttpTransport
url = 'http://127.0.0.1:8888/v1.1'
version = 1.1

118
tests/unit/queues/v1/test_message.py
View File

@@ -1,118 +0,0 @@
# Copyright (c) 2013 Rackspace, 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
from unittest import mock
from zaqarclient.queues.v1 import iterator as iterate
from zaqarclient.queues.v1 import message
from zaqarclient.tests.queues import base
from zaqarclient.tests.queues import messages as test_message
from zaqarclient.transport import http
from zaqarclient.transport import response
class TestMessageIterator(base.QueuesTestBase):
def test_no_next_iteration(self):
messages = {'links': [],
'messages': [{
'href': '/v1/queues/mine/messages/123123423',
'ttl': 800,
'age': 790,
'body': {'event': 'ActivateAccount',
'mode': 'active'}
}]
}
iterator = iterate._Iterator(self.queue.client,
messages,
'messages',
message.create_object(self.queue))
iterated = [msg for msg in iterator]
self.assertEqual(1, len(iterated))
def test_stream(self):
messages = {'links': [],
'messages': [{
'href': '/v1/queues/mine/messages/123123423',
'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(messages))
send_method.return_value = resp
# NOTE(flaper87): The first iteration will return 1 message
# and then call `_next_page` which will use the rel-next link
# to get a new set of messages.
link = {'rel': 'next',
'href': "/v1/queues/mine/messages?marker=6244-244224-783"}
messages['links'].append(link)
iterator = iterate._Iterator(self.queue.client,
messages,
'messages',
message.create_object(self.queue))
iterated = [msg for msg in iterator.stream()]
self.assertEqual(2, len(iterated))
def test_iterator_respect_paging(self):
messages = {'links': [],
'messages': [{
'href': '/v1/queues/mine/messages/123123423',
'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(messages))
send_method.return_value = resp
link = {'rel': 'next',
'href': "/v1/queues/mine/messages?marker=6244-244224-783"}
messages['links'].append(link)
iterator = iterate._Iterator(self.queue.client,
messages,
'messages',
message.create_object(self.queue))
iterated = [msg for msg in iterator]
self.assertEqual(1, len(iterated))
class QueuesV1MessageHttpUnitTest(test_message.QueuesV1MessageUnitTest):
transport_cls = http.HttpTransport
url = 'http://127.0.0.1:8888/v1'
version = 1
class QueuesV1_1MessageHttpUnitTest(test_message.QueuesV1MessageUnitTest):
transport_cls = http.HttpTransport
url = 'http://127.0.0.1:8888/v1.1'
version = 1.1

25
tests/unit/queues/v1/test_pool.py
View File

@@ -1,25 +0,0 @@
# Copyright (c) 2014 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.
from zaqarclient.tests.queues import pool
from zaqarclient.transport import http
class QueuesV1PoolHttpUnitTest(pool.QueuesV1PoolUnitTest):
transport_cls = http.HttpTransport
url = 'http://127.0.0.1:8888/v1'
version = 1

35
tests/unit/queues/v1/test_queues.py
View File

@@ -1,35 +0,0 @@
# 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.
from zaqarclient.tests.queues import queues
from zaqarclient.transport import http
class QueuesV1QueueHttpUnitTest(queues.QueuesV1QueueUnitTest):
transport_cls = http.HttpTransport
url = 'http://127.0.0.1:8888/v1'
version = 1
class QueuesV1_1QueueHttpUnitTest(queues.QueuesV1_1QueueUnitTest):
transport_cls = http.HttpTransport
url = 'http://127.0.0.1:8888/v1.1'
version = 1.1
def test_queue_exists(self):
pass

234
tests/unit/queues/v2/test_core.py
View File

@@ -13,8 +13,236 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from tests.unit.queues.v1 import test_core
import json
from unittest import mock
from zaqarclient.queues.v2 import core
from zaqarclient.tests import base
from zaqarclient.tests.transport import dummy
from zaqarclient.transport import errors
from zaqarclient.transport import request
from zaqarclient.transport import response
class TestV2Core(test_core.TestV1Core):
pass
class TestV2Core(base.TestBase):
def setUp(self):
super(TestV2Core, self).setUp()
self.transport = dummy.DummyTransport(self.conf)
def test_queue_create(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
send_method.return_value = response.Response(None, None)
req = request.Request()
core.queue_create(self.transport, req, 'test')
self.assertIn('queue_name', req.params)
def test_queue_delete(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
send_method.return_value = response.Response(None, None)
req = request.Request()
core.queue_delete(self.transport, req, 'test')
self.assertIn('queue_name', req.params)
def test_queue_exists(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
send_method.return_value = response.Response(None, None)
req = request.Request()
ret = core.queue_exists(self.transport, req, 'test')
self.assertIn('queue_name', req.params)
self.assertTrue(ret)
def test_queue_exists_not_found(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
send_method.side_effect = errors.ResourceNotFound
req = request.Request()
ret = core.queue_exists(self.transport, req, 'test')
self.assertIn('queue_name', req.params)
self.assertFalse(ret)
def test_get_queue_metadata(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, '{}')
send_method.return_value = resp
req = request.Request()
core.queue_get_metadata(self.transport, req, 'test')
def test_set_queue_metadata(self):
update_data = {'some': 'data'}
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
send_method.return_value = response.Response(None, None)
req = request.Request()
core.queue_exists(self.transport, req, update_data, 'test')
self.assertIn('queue_name', req.params)
def test_queue_get_stats(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, '{}')
send_method.return_value = resp
req = request.Request()
result = core.queue_get_stats(self.transport, req, 'test')
self.assertEqual({}, result)
def test_message_post_one(self):
messages = {'ttl': 30, 'body': 'Post one!'}
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, '{}')
send_method.return_value = resp
req = request.Request()
core.message_post(self.transport, req, 'test', messages)
self.assertIn('queue_name', req.params)
self.assertEqual(messages, json.loads(req.content))
def test_message_post_many(self):
messages = [{'ttl': 30, 'body': 'Post one!'},
{'ttl': 30, 'body': 'Post two!'},
{'ttl': 30, 'body': 'Post three!'}, ]
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, '{}')
send_method.return_value = resp
req = request.Request()
core.message_post(self.transport, req, 'test', messages)
self.assertIn('queue_name', req.params)
self.assertEqual(messages, json.loads(req.content))
def test_message_list(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, '{}')
send_method.return_value = resp
req = request.Request()
core.message_list(self.transport, req, 'test')
self.assertIn('queue_name', req.params)
def test_message_list_kwargs(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, '{}')
send_method.return_value = resp
req = request.Request()
core.message_list(self.transport, req, 'test',
marker='supermarket',
echo=False, limit=10)
self.assertIn('queue_name', req.params)
self.assertIn('limit', req.params)
self.assertIn('echo', req.params)
self.assertIn('marker', req.params)
def test_message_get_many(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, '{}')
send_method.return_value = resp
req = request.Request()
ids = ['a', 'b']
core.message_get_many(self.transport, req,
'test', ids)
self.assertIn('queue_name', req.params)
self.assertIn('ids', req.params)
self.assertEqual(ids, req.params['ids'])
def test_message_get(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, '{}')
send_method.return_value = resp
req = request.Request()
core.message_get(self.transport, req,
'test', 'message_id')
def test_message_delete(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, None)
send_method.return_value = resp
req = request.Request()
core.message_delete(self.transport, req,
'test', 'message_id')
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
ids = ['a', 'b']
req = request.Request()
core.message_delete_many(self.transport, req,
'test', ids=ids)
self.assertIn('queue_name', req.params)
self.assertIn('ids', req.params)
self.assertEqual(ids, req.params['ids'])
# ADMIN API
def test_pool_create(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, None)
send_method.return_value = resp
req = request.Request()
core.pool_create(self.transport, req,
'test_pool', {'uri': 'sqlite://',
'weight': 0})
def test_pool_get(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, None)
send_method.return_value = resp
req = request.Request()
core.pool_get(self.transport, req,
'test_pool')
def test_pool_delete(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, None)
send_method.return_value = resp
req = request.Request()
core.pool_delete(self.transport, req, 'test_pool')
def test_health(self):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, None)
send_method.return_value = resp
req = request.Request()
core.health(self.transport, req)

34
tests/unit/queues/v2/test_message.py
View File

@@ -16,7 +16,7 @@
import json
from unittest import mock
from zaqarclient.queues.v1 import iterator as iterate
from zaqarclient.queues.v2 import iterator
from zaqarclient.queues.v2 import message
from zaqarclient.tests.queues import base
from zaqarclient.tests.queues import messages as test_message
@@ -36,11 +36,11 @@ class TestMessageIterator(base.QueuesTestBase):
}]
}
iterator = iterate._Iterator(self.queue.client,
messages,
'messages',
message.create_object(self.queue))
iterated = [msg for msg in iterator]
msg_iterator = iterator._Iterator(self.queue.client,
messages,
'messages',
message.create_object(self.queue))
iterated = [msg for msg in msg_iterator]
self.assertEqual(len(iterated), 1)
def test_stream(self):
@@ -67,11 +67,12 @@ class TestMessageIterator(base.QueuesTestBase):
'href': "/v2/queues/mine/messages?marker=6244-244224-783"}
messages['links'].append(link)
iterator = iterate._Iterator(self.queue.client,
messages,
'messages',
message.create_object(self.queue))
iterated = [msg for msg in iterator.stream()]
msg_iterator = iterator._Iterator(
self.queue.client,
messages,
'messages',
message.create_object(self.queue))
iterated = [msg for msg in msg_iterator.stream()]
self.assertEqual(len(iterated), 2)
def test_iterator_respect_paging(self):
@@ -95,11 +96,12 @@ class TestMessageIterator(base.QueuesTestBase):
'href': "/v2/queues/mine/messages?marker=6244-244224-783"}
messages['links'].append(link)
iterator = iterate._Iterator(self.queue.client,
messages,
'messages',
message.create_object(self.queue))
iterated = [msg for msg in iterator]
msg_iterator = iterator._Iterator(
self.queue.client,
messages,
'messages',
message.create_object(self.queue))
iterated = [msg for msg in msg_iterator]
self.assertEqual(len(iterated), 1)

13
tests/unit/transport/test_request.py
View File

@@ -16,7 +16,6 @@
import json
from zaqarclient.queues.v1 import api as api_v1
from zaqarclient.queues.v2 import api as api_v2
from zaqarclient.tests import base
from zaqarclient.transport import request
@@ -49,18 +48,6 @@ class TestRequest(base.TestBase):
def test_request_with_right_version(self):
auth_opts = self.conf.get('auth_opts', {})
api_version = 1
req = request.prepare_request(auth_opts, api=api_version)
self.assertIsInstance(req.api, api_v1.V1)
api_version = 1.0
req = request.prepare_request(auth_opts, api=api_version)
self.assertIsInstance(req.api, api_v1.V1)
api_version = 1.1
req = request.prepare_request(auth_opts, api=api_version)
self.assertIsInstance(req.api, api_v1.V1_1)
api_version = 2
req = request.prepare_request(auth_opts, api=api_version)
self.assertIsInstance(req.api, api_v2.V2)

8
zaqarclient/queues/client.py
View File

@@ -69,20 +69,16 @@ anything you can do with this client instance can be done by accessing
the underlying API, although not recommended.
"""
from zaqarclient import errors
from zaqarclient.queues.v1 import client as cv1
from zaqarclient.queues.v2 import client as cv2
_CLIENTS = {1: cv1.Client,
1.1: cv1.Client,
2: cv2.Client}
_CLIENTS = {2: cv2.Client}
def Client(url=None, version=None, conf=None, session=None):
# NOTE: Please don't mix use the Client object with different version at
# the same time. Because the cache mechanism of queue's metadata will lead
# to unexpected response value.
# Please see zaqarclient.queues.v1.queues.Queue.metadata and
# zaqarclient.queues.v2.queues.Queue.metadata for more detail.
# Please see zaqarclient.queues.v2.queues.Queue.metadata for more detail.
try:
return _CLIENTS[version](url=url, version=version, conf=conf,
session=session)

0
zaqarclient/queues/v1/__init__.py
View File

340
zaqarclient/queues/v1/api.py
View File

@@ -1,340 +0,0 @@
# 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.
from zaqarclient.transport import api
class V1(api.Api):
label = 'v1'
schema = {
'queue_list': {
'ref': 'queues',
'method': 'GET',
'properties': {
'marker': {'type': 'string'},
'limit': {'type': 'integer'},
'detailed': {'type': 'boolean'}
}
},
'queue_create': {
'ref': 'queues/{queue_name}',
'method': 'PUT',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
},
},
'queue_exists': {
'ref': 'queues/{queue_name}',
'method': 'HEAD',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
}
},
'queue_delete': {
'ref': 'queues/{queue_name}',
'method': 'DELETE',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
}
},
'queue_set_metadata': {
'ref': 'queues/{queue_name}/metadata',
'method': 'PUT',
'required': ['queue_name'],
'properties': {
# NOTE(flaper87): Metadata is part
# of the request content. No need to
# add it here.
'queue_name': {'type': 'string'},
}
},
'queue_get_metadata': {
'ref': 'queues/{queue_name}/metadata',
'method': 'GET',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
}
},
'queue_get_stats': {
'ref': 'queues/{queue_name}/stats',
'method': 'GET',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
}
},
'message_list': {
'ref': 'queues/{queue_name}/messages',
'method': 'GET',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'},
'marker': {'type': 'string'},
'limit': {'type': 'integer'},
'echo': {'type': 'boolean'},
'include_claimed': {'type': 'boolean'},
}
},
'message_post': {
'ref': 'queues/{queue_name}/messages',
'method': 'POST',
'required': ['queue_name', 'message_id'],
'properties': {
'queue_name': {'type': 'string'},
'claim_id': {'type': 'string'},
}
},
'message_get': {
'ref': 'queues/{queue_name}/messages/{message_id}',
'method': 'GET',
'required': ['queue_name', 'message_id'],
'properties': {
'queue_name': {'type': 'string'},
'message_id': {'type': 'string'},
'claim_id': {'type': 'string'},
}
},
'message_get_many': {
'ref': 'queues/{queue_name}/messages',
'method': 'GET',
'required': ['queue_name', 'ids'],
'properties': {
'queue_name': {'type': 'string'},
'ids': {'type': 'string'},
'claim_id': {'type': 'string'},
}
},
'message_delete': {
'ref': 'queues/{queue_name}/messages/{message_id}',
'method': 'DELETE',
'required': ['queue_name', 'message_id'],
'properties': {
'queue_name': {'type': 'string'},
'message_id': {'type': 'string'},
'claim_id': {'type': 'string'},
}
},
'message_delete_many': {
'ref': 'queues/{queue_name}/messages',
'method': 'DELETE',
'required': ['queue_name', 'ids'],
'properties': {
'queue_name': {'type': 'string'},
'ids': {'type': 'string'},
}
},
'pool_create': {
'ref': 'pools/{pool_name}',
'method': 'PUT',
'required': ['pool_name'],
'properties': {
'pool_name': {'type': 'string'},
}
},
'pool_get': {
'ref': 'pools/{pool_name}',
'method': 'GET',
'required': ['pool_name'],
'properties': {
'pool_name': {'type': 'string'},
}
},
'pool_update': {
'ref': 'pools/{pool_name}',
'method': 'PATCH',
'required': ['pool_name'],
'properties': {
'pool_name': {'type': 'string'}
}
},
'pool_list': {
'ref': 'pools',
'method': 'GET',
'properties': {
'pool_name': {'type': 'string'},
'marker': {'type': 'string'},
'limit': {'type': 'integer'},
'detailed': {'type': 'boolean'}
}
},
'pool_delete': {
'ref': 'pools/{pool_name}',
'method': 'DELETE',
'required': ['pool_name'],
'properties': {
'pool_name': {'type': 'string'},
}
},
'flavor_create': {
'ref': 'flavors/{flavor_name}',
'method': 'PUT',
'required': ['flavor_name'],
'properties': {
'flavor_name': {'type': 'string'},
}
},
'flavor_get': {
'ref': 'flavors/{flavor_name}',
'method': 'GET',
'required': ['flavor_name'],
'properties': {
'flavor_name': {'type': 'string'},
}
},
'flavor_delete': {
'ref': 'flavors/{flavor_name}',
'method': 'DELETE',
'required': ['flavor_name'],
'properties': {
'flavor_name': {'type': 'string'},
}
},
'claim_create': {
'ref': 'queues/{queue_name}/claims',
'method': 'POST',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'},
'limit': {'type': 'integer'},
'grace': {'type': 'integer'}
}
},
'claim_get': {
'ref': 'queues/{queue_name}/claims/{claim_id}',
'method': 'GET',
'required': ['queue_name', 'claim_id'],
'properties': {
'queue_name': {'type': 'string'},
'claim_id': {'type': 'string'}
}
},
'claim_update': {
'ref': 'queues/{queue_name}/claims/{claim_id}',
'method': 'PATCH',
'required': ['queue_name', 'claim_id'],
'properties': {
'queue_name': {'type': 'string'},
'claim_id': {'type': 'string'}
}
},
'claim_delete': {
'ref': 'queues/{queue_name}/claims/{claim_id}',
'method': 'DELETE',
'required': ['queue_name', 'claim_id'],
'properties': {
'queue_name': {'type': 'string'},
'claim_id': {'type': 'string'}
}
},
'health': {
'admin': True,
'ref': 'health',
'method': 'GET',
},
}
class V1_1(V1):
label = 'v1.1'
schema = V1.schema.copy()
V1_1.schema.update({
'queue_get': {
'ref': 'queues/{queue_name}',
'method': 'GET',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
}
},
'queue_update': {
'ref': 'queues/{queue_name}',
'method': 'PATCH',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
}
},
'message_pop': {
'ref': 'queues/{queue_name}/messages',
'method': 'DELETE',
'required': ['queue_name', 'pop'],
'properties': {
'queue_name': {'type': 'string'},
'pop': {'type': 'integer'},
}
},
'flavor_update': {
'ref': 'flavors/{flavor_name}',
'method': 'PATCH',
'required': ['flavor_name'],
'properties': {
'flavor_name': {'type': 'string'}
}
},
'flavor_list': {
'ref': 'flavors',
'method': 'GET',
'properties': {
'flavor_name': {'type': 'string'},
'marker': {'type': 'string'},
'limit': {'type': 'integer'},
'detailed': {'type': 'boolean'}
}
},
})
del V1_1.schema['queue_get_metadata']
del V1_1.schema['queue_set_metadata']

110
zaqarclient/queues/v1/claim.py
View File

@@ -1,110 +0,0 @@
# Copyright (c) 2014 Rackspace, 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.
from zaqarclient.queues.v1 import core
from zaqarclient.queues.v1 import iterator as iterate
from zaqarclient.queues.v1 import message
class Claim(object):
def __init__(self, queue, id=None,
ttl=None, grace=None, limit=None):
self._queue = queue
self.id = id
self._ttl = ttl
self._grace = grace
self._age = None
self._limit = limit
self._message_iter = None
if id is None:
self._create()
def __repr__(self):
return '<Claim id:{id} ttl:{ttl} age:{age}>'.format(id=self.id,
ttl=self.ttl,
age=self.age)
def _get(self):
req, trans = self._queue.client._request_and_transport()
claim_res = core.claim_get(trans, req, self._queue._name,
self.id)
self._age = claim_res['age']
self._ttl = claim_res['ttl']
self._grace = claim_res.get('grace')
msgs = claim_res.get('messages', [])
self._message_iter = iterate._Iterator(self._queue.client,
msgs,
'messages',
message.create_object(
self._queue
))
def _create(self):
req, trans = self._queue.client._request_and_transport()
msgs = core.claim_create(trans, req,
self._queue._name,
ttl=self._ttl,
grace=self._grace,
limit=self._limit)
# extract the id from the first message
if msgs is not None:
if self._queue.client.api_version >= 1.1:
msgs = msgs['messages']
self.id = msgs[0]['href'].split('=')[-1]
self._message_iter = iterate._Iterator(self._queue.client,
msgs or [],
'messages',
message.create_object(
self._queue
))
def __iter__(self):
if self._message_iter is None:
self._get()
return self._message_iter
@property
def age(self):
self._get()
return self._age
@property
def ttl(self):
if self._ttl is None:
self._get()
return self._ttl
def delete(self):
req, trans = self._queue.client._request_and_transport()
core.claim_delete(trans, req, self._queue._name, self.id)
def update(self, ttl=None, grace=None):
req, trans = self._queue.client._request_and_transport()
kwargs = {}
if ttl is not None:
kwargs['ttl'] = ttl
if grace is not None:
kwargs['grace'] = grace
res = core.claim_update(trans, req, self._queue._name, self.id,
**kwargs)
# if the update succeeds, update our attributes.
if ttl is not None:
self._ttl = ttl
if grace is not None:
self._grace = grace
return res

198
zaqarclient/queues/v1/client.py
View File

@@ -1,198 +0,0 @@
# 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.
from oslo_utils import uuidutils
from zaqarclient.common import decorators
from zaqarclient.queues.v1 import core
from zaqarclient.queues.v1 import flavor
from zaqarclient.queues.v1 import iterator
from zaqarclient.queues.v1 import pool
from zaqarclient.queues.v1 import queues
from zaqarclient import transport
from zaqarclient.transport import errors
from zaqarclient.transport import request
class Client(object):
"""Client base class
:param url: Zaqar's instance base url.
:type url: str
:param version: API Version pointing to.
:type version: `int`
:param conf: CONF object.
:type conf: `oslo_config.cfg.CONF`
:param session: keystone session. But it's just place holder, we wont'
support it in v1.
"""
queues_module = queues
def __init__(self, url=None, version=1, conf=None, session=None):
self.conf = conf or {}
self.api_url = url
self.api_version = version
self.auth_opts = self.conf.get('auth_opts', {})
self.client_uuid = self.conf.get('client_uuid',
uuidutils.generate_uuid(dashed=False))
self.session = session
def _get_transport(self, request):
"""Gets a transport and caches its instance
This method gets a transport instance based on
the request's endpoint and caches that for later
use. The transport instance is invalidated whenever
a session expires.
:param request: The request to use to load the
transport instance.
:type request: :class:`zaqarclient.transport.request.Request`
"""
return transport.get_transport_for(request,
version=self.api_version,
options=self.conf)
def _request_and_transport(self):
req = request.prepare_request(self.auth_opts,
endpoint=self.api_url,
api=self.api_version,
session=self.session)
req.headers['Client-ID'] = self.client_uuid
trans = self._get_transport(req)
return req, trans
def transport(self):
"""Gets a transport based the api url and version.
:rtype: :class:`zaqarclient.transport.base.Transport`
"""
return transport.get_transport_for(self.api_url,
version=self.api_version)
def queue(self, ref, **kwargs):
"""Returns a queue instance
:param ref: Queue's reference id.
:type ref: str
:returns: A queue instance
:rtype: `queues.Queue`
"""
return self.queues_module.Queue(self, ref, **kwargs)
def queues(self, **params):
"""Gets a list of queues from the server
:returns: A list of queues
:rtype: `list`
"""
req, trans = self._request_and_transport()
queue_list = core.queue_list(trans, req, **params)
return iterator._Iterator(self,
queue_list,
'queues',
self.queues_module.create_object(self))
def follow(self, ref):
"""Follows ref.
This method instanciates a new request instance and requests
`ref`. It is intended to be used to follow a reference href
gotten from `links` sections in responses like queues' lists.
:params ref: The reference path.
:type ref: str
"""
req, trans = self._request_and_transport()
req.ref = ref
return trans.send(req).deserialized_content
# ADMIN API
def pool(self, ref, **kwargs):
"""Returns a pool instance
:param ref: Pool's reference name.
:type ref: str
:returns: A pool instance
:rtype: `pool.Pool`
"""
return pool.Pool(self, ref, **kwargs)
def pools(self, **params):
"""Gets a list of pools from the server
:param params: Filters to use for getting pools
:type params: dict.
:returns: A list of pools
:rtype: `list`
"""
req, trans = self._request_and_transport()
pool_list = core.pool_list(trans, req, **params)
return iterator._Iterator(self,
pool_list,
'pools',
pool.create_object(self))
@decorators.version(min_version=1.1)
def flavor(self, ref, **kwargs):
"""Returns a flavor instance
:param ref: Flavor's reference name.
:type ref: str
:returns: A flavor instance
:rtype: `flavor.Flavor`
"""
return flavor.Flavor(self, ref, **kwargs)
@decorators.version(min_version=1.1)
def flavors(self, **params):
"""Gets a list of flavors from the server
:param params: Filters to use for getting flavors
:type params: dict.
:returns: A list of flavors
:rtype: `list`
"""
req, trans = self._request_and_transport()
flavor_list = core.flavor_list(trans, req, **params)
return iterator._Iterator(self,
flavor_list,
'flavors',
flavor.create_object(self))
def health(self):
"""Gets the health status of Zaqar server."""
req, trans = self._request_and_transport()
try:
core.health(trans, req)
return True
except errors.ServiceUnavailableError:
return False

656
zaqarclient/queues/v1/core.py
View File

@@ -1,656 +0,0 @@
# 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.
"""
This module defines a lower level API for queues' v1. This level of the
API is responsible for packing up the final request, sending it to the server
and handling asynchronous requests.
Functions present in this module assume that:
1. The transport instance is ready to `send` the
request to the server.
2. Transport instance holds the conf instance to use for this
request.
"""
import json
import zaqarclient.transport.errors as errors
def _common_queue_ops(operation, transport, request, name, callback=None):
"""Function for common operation
This is a lower level call to get a single
instance of queue.
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param name: Queue reference name.
:type name: str
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = operation
request.params['queue_name'] = name
resp = transport.send(request)
return resp.deserialized_content
def queue_create(transport, request, name,
metadata=None, callback=None):
"""Creates a queue
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param name: Queue reference name.
:type name: str
:param metadata: Queue's metadata object. (>=v1.1)
:type metadata: `dict`
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = 'queue_create'
request.params['queue_name'] = name
request.content = metadata and json.dumps(metadata)
resp = transport.send(request)
return resp.deserialized_content
def queue_update(transport, request, name, metadata, callback=None):
"""Updates a queue's metadata using PATCH. API v1.1+ only
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param name: Queue reference name.
:type name: str
:param metadata: Queue's metadata object. (>=v1.1)
:type metadata: `dict`
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = 'queue_update'
request.params['queue_name'] = name
request.content = json.dumps(metadata)
resp = transport.send(request)
return resp.deserialized_content
def queue_exists(transport, request, name, callback=None):
"""Checks if the queue exists."""
try:
_common_queue_ops('queue_exists', transport,
request, name, callback=callback)
return True
except errors.ResourceNotFound:
return False
def queue_get(transport, request, name, callback=None):
"""Retrieve a queue."""
return _common_queue_ops('queue_get', transport,
request, name, callback=callback)
def queue_get_metadata(transport, request, name, callback=None):
"""Gets queue metadata."""
return _common_queue_ops('queue_get_metadata', transport,
request, name, callback=callback)
def queue_set_metadata(transport, request, name, metadata, callback=None):
"""Sets queue metadata."""
request.operation = 'queue_set_metadata'
request.params['queue_name'] = name
request.content = json.dumps(metadata)
transport.send(request)
def queue_get_stats(transport, request, name):
return _common_queue_ops('queue_get_stats', transport,
request, name)
def queue_delete(transport, request, name, callback=None):
"""Deletes queue."""
return _common_queue_ops('queue_delete', transport,
request, name, callback=callback)
def queue_list(transport, request, callback=None, **kwargs):
"""Gets a list of queues
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
:param kwargs: Optional arguments for this operation.
- marker: Where to start getting queues from.
- limit: Maximum number of queues to get.
"""
request.operation = 'queue_list'
request.params.update(kwargs)
resp = transport.send(request)
if not resp.content:
return {'links': [], 'queues': []}
return resp.deserialized_content
def message_list(transport, request, queue_name, callback=None, **kwargs):
"""Gets a list of messages in queue `queue_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param queue_name: Queue reference name.
:type queue_name: str
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
:param kwargs: Optional arguments for this operation.
- marker: Where to start getting messages from.
- limit: Maximum number of messages to get.
- echo: Whether to get our own messages.
- include_claimed: Whether to include claimed
messages.
"""
request.operation = 'message_list'
request.params['queue_name'] = queue_name
# NOTE(flaper87): Assume passed params
# are accepted by the API, if not, the
# API itself will raise an error.
request.params.update(kwargs)
resp = transport.send(request)
if not resp.content:
# NOTE(flaper87): We could also return None
# or an empty dict, however, we're giving
# more value to a consistent API here by
# returning a compliant dict with empty
# `links` and `messages`
return {'links': [], 'messages': []}
return resp.deserialized_content
def message_post(transport, request, queue_name, messages, callback=None):
"""Post messages to `queue_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param queue_name: Queue reference name.
:type queue_name: str
:param messages: One or more messages to post.
:param messages: `list`
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = 'message_post'
request.params['queue_name'] = queue_name
request.content = json.dumps(messages)
resp = transport.send(request)
return resp.deserialized_content
def message_get(transport, request, queue_name, message_id, callback=None):
"""Gets one message from the queue by id
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param queue_name: Queue reference name.
:type queue_name: str
:param message_id: Message reference.
:param message_id: str
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = 'message_get'
request.params['queue_name'] = queue_name
request.params['message_id'] = message_id
resp = transport.send(request)
return resp.deserialized_content
def message_get_many(transport, request, queue_name, messages, callback=None):
"""Gets many messages by id
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param queue_name: Queue reference name.
:type queue_name: str
:param messages: Messages references.
:param messages: list of str
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = 'message_get_many'
request.params['queue_name'] = queue_name
request.params['ids'] = messages
resp = transport.send(request)
return resp.deserialized_content
def message_delete(transport, request, queue_name, message_id,
claim_id=None, callback=None):
"""Deletes messages from `queue_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param queue_name: Queue reference name.
:type queue_name: str
:param message_id: Message reference.
:param message_id: str
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = 'message_delete'
request.params['queue_name'] = queue_name
request.params['message_id'] = message_id
if claim_id:
request.params['claim_id'] = claim_id
transport.send(request)
def message_delete_many(transport, request, queue_name,
ids, callback=None):
"""Deletes `ids` messages from `queue_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param queue_name: Queue reference name.
:type queue_name: str
:param ids: Ids of the messages to delete
:type ids: List of str
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = 'message_delete_many'
request.params['queue_name'] = queue_name
request.params['ids'] = ids
transport.send(request)
def message_pop(transport, request, queue_name,
count, callback=None):
"""Pops out `count` messages from `queue_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param queue_name: Queue reference name.
:type queue_name: str
:param count: Number of messages to pop.
:type count: int
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = 'message_delete_many'
request.params['queue_name'] = queue_name
request.params['pop'] = count
resp = transport.send(request)
return resp.deserialized_content
def claim_create(transport, request, queue_name, **kwargs):
"""Creates a Claim `claim_id` on the queue `queue_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
"""
request.operation = 'claim_create'
request.params['queue_name'] = queue_name
if 'limit' in kwargs:
request.params['limit'] = kwargs.pop('limit')
request.content = json.dumps(kwargs)
resp = transport.send(request)
return resp.deserialized_content
def claim_get(transport, request, queue_name, claim_id):
"""Gets a Claim `claim_id`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
"""
request.operation = 'claim_get'
request.params['queue_name'] = queue_name
request.params['claim_id'] = claim_id
resp = transport.send(request)
return resp.deserialized_content
def claim_update(transport, request, queue_name, claim_id, **kwargs):
"""Updates a Claim `claim_id`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
"""
request.operation = 'claim_update'
request.params['queue_name'] = queue_name
request.params['claim_id'] = claim_id
request.content = json.dumps(kwargs)
resp = transport.send(request)
return resp.deserialized_content
def claim_delete(transport, request, queue_name, claim_id):
"""Deletes a Claim `claim_id`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
"""
request.operation = 'claim_delete'
request.params['queue_name'] = queue_name
request.params['claim_id'] = claim_id
transport.send(request)
def pool_get(transport, request, pool_name, callback=None):
"""Gets pool data
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param pool_name: Pool reference name.
:type pool_name: str
"""
request.operation = 'pool_get'
request.params['pool_name'] = pool_name
resp = transport.send(request)
return resp.deserialized_content
def pool_create(transport, request, pool_name, pool_data):
"""Creates a pool called `pool_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param pool_name: Pool reference name.
:type pool_name: str
:param pool_data: Pool's properties, i.e: weight, uri, options.
:type pool_data: `dict`
"""
request.operation = 'pool_create'
request.params['pool_name'] = pool_name
request.content = json.dumps(pool_data)
transport.send(request)
def pool_update(transport, request, pool_name, pool_data):
"""Updates the pool `pool_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param pool_name: Pool reference name.
:type pool_name: str
:param pool_data: Pool's properties, i.e: weight, uri, options.
:type pool_data: `dict`
"""
request.operation = 'pool_update'
request.params['pool_name'] = pool_name
request.content = json.dumps(pool_data)
resp = transport.send(request)
return resp.deserialized_content
def pool_list(transport, request, **kwargs):
"""Gets a list of pools
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param kwargs: Optional arguments for this operation.
- marker: Where to start getting pools from.
- limit: Maximum number of pools to get.
"""
request.operation = 'pool_list'
request.params.update(kwargs)
resp = transport.send(request)
if not resp.content:
return {'links': [], 'pools': []}
return resp.deserialized_content
def pool_delete(transport, request, pool_name):
"""Deletes the pool `pool_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param pool_name: Pool reference name.
:type pool_name: str
"""
request.operation = 'pool_delete'
request.params['pool_name'] = pool_name
transport.send(request)
def flavor_create(transport, request, name, flavor_data):
"""Creates a flavor called `name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param name: Flavor reference name.
:type name: str
:param flavor_data: Flavor's properties, i.e: pool, capabilities.
:type flavor_data: `dict`
"""
request.operation = 'flavor_create'
request.params['flavor_name'] = name
request.content = json.dumps(flavor_data)
transport.send(request)
def flavor_get(transport, request, flavor_name, callback=None):
"""Gets flavor data
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param flavor_name: Flavor reference name.
:type flavor_name: str
"""
request.operation = 'flavor_get'
request.params['flavor_name'] = flavor_name
resp = transport.send(request)
return resp.deserialized_content
def flavor_update(transport, request, flavor_name, flavor_data):
"""Updates the flavor `flavor_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param flavor_name: Flavor reference name.
:type flavor_name: str
:param flavor_data: Flavor's properties, i.e: pool, capabilities.
:type flavor_data: `dict`
"""
request.operation = 'flavor_update'
request.params['flavor_name'] = flavor_name
request.content = json.dumps(flavor_data)
resp = transport.send(request)
return resp.deserialized_content
def flavor_list(transport, request, **kwargs):
"""Gets a list of flavors
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param kwargs: Optional arguments for this operation.
- marker: Where to start getting flavors from.
- limit: Maximum number of flavors to get.
"""
request.operation = 'flavor_list'
request.params.update(kwargs)
resp = transport.send(request)
if not resp.content:
return {'links': [], 'flavors': []}
return resp.deserialized_content
def flavor_delete(transport, request, name):
"""Deletes the flavor `name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param name: Flavor reference name.
:type name: str
"""
request.operation = 'flavor_delete'
request.params['flavor_name'] = name
transport.send(request)
def health(transport, request, callback=None):
"""Check the health of web head for load balancing
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = 'health'
resp = transport.send(request)
return resp.deserialized_content

72
zaqarclient/queues/v1/flavor.py
View File

@@ -1,72 +0,0 @@
# Copyright (c) 2014 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.
from zaqarclient.queues.v1 import core
class Flavor(object):
def __init__(self, client, name,
pool_list=None, auto_create=True,
**kwargs):
self.client = client
self.name = name
self.pool_list = pool_list
self.capabilities = kwargs.get('capabilities', {})
if auto_create:
self.ensure_exists()
def ensure_exists(self):
"""Ensures pool exists
This method is not race safe,
the pool could've been deleted
right after it was called.
"""
req, trans = self.client._request_and_transport()
# As of now on PUT, zaqar server updates flavor if it is already
# exists else it will create a new one. The zaqar client should
# maitain symmetry with zaqar server.
# TBD(mdnadeem): Have to change this code when zaqar server
# behaviour change for PUT operation.
data = {'pool_list': self.pool_list}
if self.client.api_version <= 1.1:
data['capabilities'] = self.capabilities
req, trans = self.client._request_and_transport()
core.flavor_create(trans, req, self.name, data)
def update(self, flavor_data):
req, trans = self.client._request_and_transport()
core.flavor_update(trans, req, self.name, flavor_data)
for key, value in flavor_data.items():
setattr(self, key, value)
def delete(self):
req, trans = self.client._request_and_transport()
core.flavor_delete(trans, req, self.name)
def get(self):
req, trans = self.client._request_and_transport()
return core.flavor_get(trans, req, self.name, callback=None)
def create_object(parent):
return lambda kwargs: Flavor(parent, kwargs.pop('name'),
auto_create=False, **kwargs)

65
zaqarclient/queues/v1/message.py
View File

@@ -1,65 +0,0 @@
# 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.
"""Implements a message controller that understands Zaqar messages."""
from zaqarclient.queues.v1 import core
class Message(object):
"""A handler for Zaqar server Message resources.
Attributes are only downloaded once - at creation time.
"""
def __init__(self, queue, ttl, age, body, href=None, id=None,
claim_id=None, claim_count=0, checksum=None):
self.queue = queue
self.href = href
self.ttl = ttl
self.age = age
self.body = body
self.claim_count = claim_count
self.checksum = checksum
# NOTE(flaper87): Is this really
# necessary? Should this be returned
# by Zaqar?
# The url has two forms depending on if it has been claimed.
# /v1/queues/worker-jobs/messages/5c6939a8?claim_id=63c9a592
# or
# /v1/queues/worker-jobs/messages/5c6939a8
if id is None:
self._id = href.split('/')[-1]
if '?' in self._id:
self._id = self._id.split('?')[0]
else:
self._id = id
def __repr__(self):
return '<Message id:{id} ttl:{ttl}>'.format(id=self._id,
ttl=self.ttl)
@property
def claim_id(self):
if '=' in self.href:
return self.href.split('=')[-1]
def delete(self):
req, trans = self.queue.client._request_and_transport()
core.message_delete(trans, req, self.queue._name,
self._id, self.claim_id)
def create_object(parent):
return lambda args: Message(parent, **args)

77
zaqarclient/queues/v1/pool.py
View File

@@ -1,77 +0,0 @@
# Copyright (c) 2014 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.
from zaqarclient.queues.v1 import core
class Pool(object):
def __init__(self, client, name,
weight=None, uri=None,
flavor=None, auto_create=True,
**kwargs):
self.client = client
self.uri = uri
self.name = name
self.weight = weight
self.flavor = flavor
self.options = kwargs.get("options", {})
if auto_create:
self.ensure_exists()
def ensure_exists(self):
"""Ensures pool exists
This method is not race safe,
the pool could've been deleted
right after it was called.
"""
req, trans = self.client._request_and_transport()
# As of now on PUT, zaqar server updates pool if it is already
# exists else it will create a new one. The zaqar client should
# maitain symmetry with zaqar server.
# TBD(mdnadeem): Have to change this code when zaqar server
# behaviour change for PUT operation.
data = {'uri': self.uri,
'weight': self.weight,
'options': self.options}
if self.client.api_version >= 1.1 and self.flavor:
data['flavor'] = self.flavor
req, trans = self.client._request_and_transport()
core.pool_create(trans, req, self.name, data)
def update(self, pool_data):
req, trans = self.client._request_and_transport()
core.pool_update(trans, req, self.name, pool_data)
for key, value in pool_data.items():
setattr(self, key, value)
def delete(self):
req, trans = self.client._request_and_transport()
core.pool_delete(trans, req, self.name)
def get(self):
req, trans = self.client._request_and_transport()
return core.pool_get(trans, req, self.name, callback=None)
def create_object(parent):
return lambda kwargs: Pool(parent, kwargs.pop('name'),
auto_create=False, **kwargs)

262
zaqarclient/queues/v1/queues.py
View File

@@ -1,262 +0,0 @@
# 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 re
from zaqarclient._i18n import _ # noqa
from zaqarclient import errors
from zaqarclient.queues.v1 import claim as claim_api
from zaqarclient.queues.v1 import core
from zaqarclient.queues.v1 import iterator
from zaqarclient.queues.v1 import message
# NOTE(wanghao): This is copied from Zaqar server side, so if server have
# updated it someday, we should update it here to keep consistent.
QUEUE_NAME_REGEX = re.compile(r'^[a-zA-Z0-9_\-]+$')
class Queue(object):
message_module = message
def __init__(self, client, name, href=None, metadata=None,
auto_create=True, force_create=False):
"""Initialize queue object
:param client: The client object of Zaqar.
:type client: `object`
:param name: Name of the queue.
:type name: str
:param href : Hypertext Reference.
:type href: str
:param metadata : A metadata object of the queue.
:type metadata: `dict`
:param auto_create: If create the queue automatically in database.
:type auto_create: `boolean`
:param force_create: If create the queue and skip the API version
check, which is useful for command line interface.
:type force_create: `boolean`
:returns: The queue object.
"""
self.client = client
if name == "":
raise ValueError(_('Queue name does not have a value'))
if not QUEUE_NAME_REGEX.match(str(name)):
raise ValueError(_('The queue name may only contain ASCII '
'letters, digits, underscores and dashes.'))
# NOTE(flaper87) Queue Info
self._name = name
self._metadata = metadata
self._href = href
# NOTE(flwang): If force_create is True, then even though auto_create
# is not True, the queue should be created anyway.
if auto_create or force_create:
self.ensure_exists(force_create=force_create)
@property
def name(self):
return self._name
@property
def href(self):
return self._href
@property
def metadata_dict(self):
return dict(self.metadata())
def exists(self):
"""Checks if the queue exists."""
req, trans = self.client._request_and_transport()
if self.client.api_version >= 1.1:
raise errors.InvalidOperation("Unavailable on versions >= 1.1")
else:
return core.queue_exists(trans, req, self._name)
def ensure_exists(self, force_create=False):
"""Ensures a queue exists
This method is not race safe,
the queue could've been deleted
right after it was called.
"""
req, trans = self.client._request_and_transport()
if force_create or self.client.api_version < 1.1:
core.queue_create(trans, req, self._name)
def metadata(self, new_meta=None, force_reload=False):
"""Get metadata and return it
:param new_meta: A dictionary containing
an updated metadata object. If present
the queue metadata will be updated in
remote server. If the new_meta is empty,
the metadata object will be cleared.
:type new_meta: `dict`
:param force_reload: Whether to ignored the
cached metadata and reload it from the
server.
:type force_reload: `bool`
:returns: The queue metadata.
"""
req, trans = self.client._request_and_transport()
# NOTE(jeffrey4l): Ensure that metadata is cleared when the new_meta
# is an empty dict.
if new_meta is not None:
if self.client.api_version == 1.1:
raise RuntimeError("V1.1 doesn't support to set the queue's "
"metadata. Please use V1.0 or V2.")
core.queue_set_metadata(trans, req, self._name, new_meta)
self._metadata = new_meta
# TODO(flaper87): Cache with timeout
if self._metadata and not force_reload:
return self._metadata
if self.client.api_version >= 1.1:
self._metadata = core.queue_get(trans, req, self._name)
else:
self._metadata = core.queue_get_metadata(trans, req, self._name)
return self._metadata
@property
def stats(self):
req, trans = self.client._request_and_transport()
return core.queue_get_stats(trans, req, self._name)
def delete(self):
req, trans = self.client._request_and_transport()
core.queue_delete(trans, req, self._name)
# Messages API
def post(self, messages):
"""Posts one or more messages to this queue
:param messages: One or more messages to post
:type messages: `list` or `dict`
:returns: A dict with the result of this operation.
:rtype: `dict`
"""
if not isinstance(messages, list):
messages = [messages]
if self.client.api_version >= 1.1:
messages = {'messages': messages}
req, trans = self.client._request_and_transport()
# TODO(flaper87): Return a list of messages
return core.message_post(trans, req,
self._name, messages)
def message(self, message_id):
"""Gets a message by id
:param message_id: Message's reference
:type message_id: str
:returns: A message
:rtype: `dict`
"""
req, trans = self.client._request_and_transport()
msg = core.message_get(trans, req, self._name,
message_id)
return self.message_module.Message(self, **msg)
def messages(self, *messages, **params):
"""Gets a list of messages from the server
This method returns a list of messages, it can be
used to retrieve a set of messages by id or to
walk through the active messages by using the
collection endpoint.
The `messages` and `params` params are mutually exclusive
and the former has the priority.
:param messages: List of messages' ids to retrieve.
:type messages: *args of str
:param params: Filters to use for getting messages
:type params: **kwargs dict.
:returns: List of messages
:rtype: `list`
"""
req, trans = self.client._request_and_transport()
# TODO(flaper87): Return a MessageIterator.
# This iterator should handle limits, pagination
# and messages deserialization.
if messages:
msgs = core.message_get_many(trans, req,
self._name, messages)
else:
# NOTE(flaper87): It's safe to access messages
# directly. If something wrong happens, the core
# API will raise the right exceptions.
msgs = core.message_list(trans, req,
self._name,
**params)
return iterator._Iterator(self.client,
msgs,
'messages',
self.message_module.create_object(self))
def delete_messages(self, *messages):
"""Deletes a set of messages from the server
:param messages: List of messages' ids to delete.
:type messages: *args of str
"""
req, trans = self.client._request_and_transport()
return core.message_delete_many(trans, req, self._name,
set(messages))
def pop(self, count=1):
"""Pop `count` messages from the server
:param count: Number of messages to pop.
:type count: int
:returns: List of messages
:rtype: `list`
"""
req, trans = self.client._request_and_transport()
msgs = core.message_pop(trans, req, self._name, count=count)
return iterator._Iterator(self.client,
msgs,
'messages',
self.message_module.create_object(self))
def claim(self, id=None, ttl=None, grace=None,
limit=None):
return claim_api.Claim(self, id=id, ttl=ttl, grace=grace, limit=limit)
def create_object(parent):
return lambda args: Queue(parent, args["name"], auto_create=False)

402
zaqarclient/queues/v2/api.py
View File

@@ -13,88 +13,338 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from zaqarclient.queues.v1 import api
from zaqarclient.transport import api
class V2(api.V1_1):
class V2(api.Api):
label = 'v2'
schema = api.V1_1.schema.copy()
V2.schema.update({
'queue_purge': {
'ref': 'queues/{queue_name}/purge',
'method': 'POST',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
}
},
'signed_url_create': {
'ref': 'queues/{queue_name}/share',
'method': 'POST',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
schema = {
'queue_list': {
'ref': 'queues',
'method': 'GET',
'properties': {
'marker': {'type': 'string'},
'limit': {'type': 'integer'},
'detailed': {'type': 'boolean'}
}
},
},
'subscription_create': {
'ref': 'queues/{queue_name}/subscriptions',
'method': 'POST',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
'queue_create': {
'ref': 'queues/{queue_name}',
'method': 'PUT',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
},
},
},
'subscription_get': {
'ref': 'queues/{queue_name}/subscriptions/{subscription_id}',
'method': 'GET',
'required': ['queue_name', 'subscription_id'],
'properties': {
'queue_name': {'type': 'string'},
'subscription_id': {'type': 'string'}
'queue_get': {
'ref': 'queues/{queue_name}',
'method': 'GET',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
}
},
'queue_exists': {
'ref': 'queues/{queue_name}',
'method': 'HEAD',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
}
},
'queue_update': {
'ref': 'queues/{queue_name}',
'method': 'PATCH',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
}
},
'queue_delete': {
'ref': 'queues/{queue_name}',
'method': 'DELETE',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
}
},
'queue_get_stats': {
'ref': 'queues/{queue_name}/stats',
'method': 'GET',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
}
},
'queue_purge': {
'ref': 'queues/{queue_name}/purge',
'method': 'POST',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
}
},
},
'subscription_update': {
'ref': 'queues/{queue_name}/subscriptions/{subscription_id}',
'method': 'PATCH',
'required': ['queue_name', 'subscription_id'],
'properties': {
'queue_name': {'type': 'string'},
'subscription_id': {'type': 'string'}
}
},
'subscription_delete': {
'ref': 'queues/{queue_name}/subscriptions/{subscription_id}',
'method': 'DELETE',
'required': ['queue_name', 'subscription_id'],
'properties': {
'queue_name': {'type': 'string'},
'subscription_id': {'type': 'string'}
}
},
'subscription_list': {
'ref': 'queues/{queue_name}/subscriptions',
'method': 'GET',
'properties': {
'marker': {'type': 'string'},
'limit': {'type': 'integer'},
'detailed': {'type': 'boolean'}
}
},
'ping': {
'ref': 'ping',
'method': 'GET',
},
'message_list': {
'ref': 'queues/{queue_name}/messages',
'method': 'GET',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'},
'marker': {'type': 'string'},
'limit': {'type': 'integer'},
'echo': {'type': 'boolean'},
'include_claimed': {'type': 'boolean'},
}
},
'message_post': {
'ref': 'queues/{queue_name}/messages',
'method': 'POST',
'required': ['queue_name', 'message_id'],
'properties': {
'queue_name': {'type': 'string'},
'claim_id': {'type': 'string'},
}
},
'message_get': {
'ref': 'queues/{queue_name}/messages/{message_id}',
'method': 'GET',
'required': ['queue_name', 'message_id'],
'properties': {
'queue_name': {'type': 'string'},
'message_id': {'type': 'string'},
'claim_id': {'type': 'string'},
}
},
'message_get_many': {
'ref': 'queues/{queue_name}/messages',
'method': 'GET',
'required': ['queue_name', 'ids'],
'properties': {
'queue_name': {'type': 'string'},
'ids': {'type': 'string'},
'claim_id': {'type': 'string'},
}
},
'message_delete': {
'ref': 'queues/{queue_name}/messages/{message_id}',
'method': 'DELETE',
'required': ['queue_name', 'message_id'],
'properties': {
'queue_name': {'type': 'string'},
'message_id': {'type': 'string'},
'claim_id': {'type': 'string'},
}
},
'message_delete_many': {
'ref': 'queues/{queue_name}/messages',
'method': 'DELETE',
'required': ['queue_name', 'ids'],
'properties': {
'queue_name': {'type': 'string'},
'ids': {'type': 'string'},
}
},
'message_pop': {
'ref': 'queues/{queue_name}/messages',
'method': 'DELETE',
'required': ['queue_name', 'pop'],
'properties': {
'queue_name': {'type': 'string'},
'pop': {'type': 'integer'},
}
},
'health': {
'ref': 'health',
'method': 'GET',
},
'pool_list': {
'ref': 'pools',
'method': 'GET',
'properties': {
'pool_name': {'type': 'string'},
'marker': {'type': 'string'},
'limit': {'type': 'integer'},
'detailed': {'type': 'boolean'}
}
},
'pool_create': {
'ref': 'pools/{pool_name}',
'method': 'PUT',
'required': ['pool_name'],
'properties': {
'pool_name': {'type': 'string'},
}
},
'pool_get': {
'ref': 'pools/{pool_name}',
'method': 'GET',
'required': ['pool_name'],
'properties': {
'pool_name': {'type': 'string'},
}
},
'pool_update': {
'ref': 'pools/{pool_name}',
'method': 'PATCH',
'required': ['pool_name'],
'properties': {
'pool_name': {'type': 'string'}
}
},
'pool_delete': {
'ref': 'pools/{pool_name}',
'method': 'DELETE',
'required': ['pool_name'],
'properties': {
'pool_name': {'type': 'string'},
}
},
'homedoc': {
'ref': '',
'method': 'GET',
},
})
'subscription_list': {
'ref': 'queues/{queue_name}/subscriptions',
'method': 'GET',
'properties': {
'marker': {'type': 'string'},
'limit': {'type': 'integer'},
'detailed': {'type': 'boolean'}
}
},
'subscription_create': {
'ref': 'queues/{queue_name}/subscriptions',
'method': 'POST',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
},
},
'subscription_get': {
'ref': 'queues/{queue_name}/subscriptions/{subscription_id}',
'method': 'GET',
'required': ['queue_name', 'subscription_id'],
'properties': {
'queue_name': {'type': 'string'},
'subscription_id': {'type': 'string'}
},
},
'subscription_update': {
'ref': 'queues/{queue_name}/subscriptions/{subscription_id}',
'method': 'PATCH',
'required': ['queue_name', 'subscription_id'],
'properties': {
'queue_name': {'type': 'string'},
'subscription_id': {'type': 'string'}
}
},
'subscription_delete': {
'ref': 'queues/{queue_name}/subscriptions/{subscription_id}',
'method': 'DELETE',
'required': ['queue_name', 'subscription_id'],
'properties': {
'queue_name': {'type': 'string'},
'subscription_id': {'type': 'string'}
}
},
'flavor_list': {
'ref': 'flavors',
'method': 'GET',
'properties': {
'flavor_name': {'type': 'string'},
'marker': {'type': 'string'},
'limit': {'type': 'integer'},
'detailed': {'type': 'boolean'}
}
},
'flavor_create': {
'ref': 'flavors/{flavor_name}',
'method': 'PUT',
'required': ['flavor_name'],
'properties': {
'flavor_name': {'type': 'string'},
}
},
'flavor_get': {
'ref': 'flavors/{flavor_name}',
'method': 'GET',
'required': ['flavor_name'],
'properties': {
'flavor_name': {'type': 'string'},
}
},
'flavor_update': {
'ref': 'flavors/{flavor_name}',
'method': 'PATCH',
'required': ['flavor_name'],
'properties': {
'flavor_name': {'type': 'string'}
}
},
'flavor_delete': {
'ref': 'flavors/{flavor_name}',
'method': 'DELETE',
'required': ['flavor_name'],
'properties': {
'flavor_name': {'type': 'string'},
}
},
'claim_create': {
'ref': 'queues/{queue_name}/claims',
'method': 'POST',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'},
'limit': {'type': 'integer'},
'grace': {'type': 'integer'}
}
},
'claim_get': {
'ref': 'queues/{queue_name}/claims/{claim_id}',
'method': 'GET',
'required': ['queue_name', 'claim_id'],
'properties': {
'queue_name': {'type': 'string'},
'claim_id': {'type': 'string'}
}
},
'claim_update': {
'ref': 'queues/{queue_name}/claims/{claim_id}',
'method': 'PATCH',
'required': ['queue_name', 'claim_id'],
'properties': {
'queue_name': {'type': 'string'},
'claim_id': {'type': 'string'}
}
},
'claim_delete': {
'ref': 'queues/{queue_name}/claims/{claim_id}',
'method': 'DELETE',
'required': ['queue_name', 'claim_id'],
'properties': {
'queue_name': {'type': 'string'},
'claim_id': {'type': 'string'}
}
},
'signed_url_create': {
'ref': 'queues/{queue_name}/share',
'method': 'POST',
'required': ['queue_name'],
'properties': {
'queue_name': {'type': 'string'}
},
},
'ping': {
'ref': 'ping',
'method': 'GET',
},
'health': {
'ref': 'health',
'method': 'GET',
},
'homedoc': {
'ref': '',
'method': 'GET',
},
}

85
zaqarclient/queues/v2/claim.py
View File

@@ -13,13 +13,44 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from zaqarclient.queues.v1 import claim
from zaqarclient.queues.v1 import iterator as iterate
from zaqarclient.queues.v2 import core
from zaqarclient.queues.v2 import iterator
from zaqarclient.queues.v2 import message
class Claim(claim.Claim):
class Claim(object):
def __init__(self, queue, id=None,
ttl=None, grace=None, limit=None):
self._queue = queue
self.id = id
self._ttl = ttl
self._grace = grace
self._age = None
self._limit = limit
self._message_iter = None
if id is None:
self._create()
def __repr__(self):
return '<Claim id:{id} ttl:{ttl} age:{age}>'.format(id=self.id,
ttl=self.ttl,
age=self.age)
def _get(self):
req, trans = self._queue.client._request_and_transport()
claim_res = core.claim_get(trans, req, self._queue._name,
self.id)
self._age = claim_res['age']
self._ttl = claim_res['ttl']
self._grace = claim_res.get('grace')
msgs = claim_res.get('messages', [])
self._message_iter = iterator._Iterator(self._queue.client,
msgs,
'messages',
message.create_object(
self._queue
))
def _create(self):
req, trans = self._queue.client._request_and_transport()
@@ -34,9 +65,45 @@ class Claim(claim.Claim):
msgs = msgs['messages']
self.id = msgs[0]['href'].split('=')[-1]
self._message_iter = iterate._Iterator(self._queue.client,
msgs or [],
'messages',
message.create_object(
self._queue
))
self._message_iter = iterator._Iterator(self._queue.client,
msgs or [],
'messages',
message.create_object(
self._queue
))
def __iter__(self):
if self._message_iter is None:
self._get()
return self._message_iter
@property
def age(self):
self._get()
return self._age
@property
def ttl(self):
if self._ttl is None:
self._get()
return self._ttl
def delete(self):
req, trans = self._queue.client._request_and_transport()
core.claim_delete(trans, req, self._queue._name, self.id)
def update(self, ttl=None, grace=None):
req, trans = self._queue.client._request_and_transport()
kwargs = {}
if ttl is not None:
kwargs['ttl'] = ttl
if grace is not None:
kwargs['grace'] = grace
res = core.claim_update(trans, req, self._queue._name, self.id,
**kwargs)
# if the update succeeds, update our attributes.
if ttl is not None:
self._ttl = ttl
if grace is not None:
self._grace = grace
return res

124
zaqarclient/queues/v2/client.py
View File

@@ -14,15 +14,17 @@
# limitations under the License.
from oslo_utils import uuidutils
from zaqarclient.common import decorators
from zaqarclient.queues.v1 import client
from zaqarclient.queues.v1 import iterator
from zaqarclient.queues.v2 import core
from zaqarclient.queues.v2 import flavor
from zaqarclient.queues.v2 import iterator
from zaqarclient.queues.v2 import pool
from zaqarclient.queues.v2 import queues
from zaqarclient.queues.v2 import subscription
from zaqarclient import transport
from zaqarclient.transport import request
class Client(client.Client):
class Client(object):
"""Client base class
:param url: Zaqar's instance base url.
@@ -45,6 +47,42 @@ class Client(client.Client):
uuidutils.generate_uuid(dashed=False))
self.session = session
def _get_transport(self, request):
"""Gets a transport and caches its instance
This method gets a transport instance based on
the request's endpoint and caches that for later
use. The transport instance is invalidated whenever
a session expires.
:param request: The request to use to load the
transport instance.
:type request: :class:`zaqarclient.transport.request.Request`
"""
return transport.get_transport_for(request,
version=self.api_version,
options=self.conf)
def _request_and_transport(self):
req = request.prepare_request(self.auth_opts,
endpoint=self.api_url,
api=self.api_version,
session=self.session)
req.headers['Client-ID'] = self.client_uuid
trans = self._get_transport(req)
return req, trans
def transport(self):
"""Gets a transport based the api url and version.
:rtype: :class:`zaqarclient.transport.base.Transport`
"""
return transport.get_transport_for(self.api_url,
version=self.api_version)
def queue(self, ref, **kwargs):
"""Returns a queue instance
@@ -74,7 +112,80 @@ class Client(client.Client):
self.queues_module.create_object(self))
return (list_iter, count)
@decorators.version(min_version=2)
def follow(self, ref):
"""Follows ref.
This method instanciates a new request instance and requests
`ref`. It is intended to be used to follow a reference href
gotten from `links` sections in responses like queues' lists.
:params ref: The reference path.
:type ref: str
"""
req, trans = self._request_and_transport()
req.ref = ref
return trans.send(req).deserialized_content
# ADMIN API
def pool(self, ref, **kwargs):
"""Returns a pool instance
:param ref: Pool's reference name.
:type ref: str
:returns: A pool instance
:rtype: `pool.Pool`
"""
return pool.Pool(self, ref, **kwargs)
def pools(self, **params):
"""Gets a list of pools from the server
:param params: Filters to use for getting pools
:type params: dict.
:returns: A list of pools
:rtype: `list`
"""
req, trans = self._request_and_transport()
pool_list = core.pool_list(trans, req, **params)
return iterator._Iterator(self,
pool_list,
'pools',
pool.create_object(self))
def flavor(self, ref, **kwargs):
"""Returns a flavor instance
:param ref: Flavor's reference name.
:type ref: str
:returns: A flavor instance
:rtype: `flavor.Flavor`
"""
return flavor.Flavor(self, ref, **kwargs)
def flavors(self, **params):
"""Gets a list of flavors from the server
:param params: Filters to use for getting flavors
:type params: dict.
:returns: A list of flavors
:rtype: `list`
"""
req, trans = self._request_and_transport()
flavor_list = core.flavor_list(trans, req, **params)
return iterator._Iterator(self,
flavor_list,
'flavors',
flavor.create_object(self))
def subscription(self, queue_name, **kwargs):
"""Returns a subscription instance
@@ -86,7 +197,6 @@ class Client(client.Client):
"""
return subscription.Subscription(self, queue_name, **kwargs)
@decorators.version(min_version=2)
def subscriptions(self, queue_name, **params):
"""Gets a list of subscriptions from the server
@@ -111,13 +221,11 @@ class Client(client.Client):
req, trans = self._request_and_transport()
return core.ping(trans, req)
@decorators.version(min_version=1.1)
def health(self):
"""Gets the detailed health status of Zaqar server."""
req, trans = self._request_and_transport()
return core.health(trans, req)
@decorators.version(min_version=1.1)
def homedoc(self):
"""Get the detailed resource doc of Zaqar server"""
req, trans = self._request_and_transport()

612
zaqarclient/queues/v2/core.py
View File

@@ -32,39 +32,60 @@ import json
from oslo_utils import timeutils
from zaqarclient.queues.v1 import core
import zaqarclient.transport.errors as errors
queue_create = core.queue_create
queue_exists = core.queue_exists
queue_get = core.queue_get
queue_get_metadata = core.queue_get_metadata
queue_set_metadata = core.queue_set_metadata
queue_get_stats = core.queue_get_stats
queue_delete = core.queue_delete
queue_list = core.queue_list
message_get = core.message_get
message_list = core.message_list
message_post = core.message_post
message_delete = core.message_delete
message_delete_many = core.message_delete_many
pool_get = core.pool_get
pool_create = core.pool_create
pool_delete = core.pool_delete
pool_update = core.pool_update
pool_list = core.pool_list
flavor_get = core.flavor_get
flavor_create = core.flavor_create
flavor_delete = core.flavor_delete
flavor_update = core.flavor_update
flavor_list = core.flavor_list
claim_create = core.claim_create
claim_get = core.claim_get
claim_update = core.claim_update
claim_delete = core.claim_delete
def _common_queue_ops(operation, transport, request, name, callback=None):
"""Function for common operation
This is a lower level call to get a single
instance of queue.
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param name: Queue reference name.
:type name: str
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = operation
request.params['queue_name'] = name
resp = transport.send(request)
return resp.deserialized_content
def queue_create(transport, request, name,
metadata=None, callback=None):
"""Creates a queue
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param name: Queue reference name.
:type name: str
:param metadata: Queue's metadata object. (>=v1.1)
:type metadata: `dict`
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = 'queue_create'
request.params['queue_name'] = name
request.content = metadata and json.dumps(metadata)
resp = transport.send(request)
return resp.deserialized_content
def queue_update(transport, request, name, metadata, callback=None):
"""Updates a queue's metadata using PATCH for API v2
"""Updates a queue's metadata using PATCH
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
@@ -88,6 +109,77 @@ def queue_update(transport, request, name, metadata, callback=None):
return resp.deserialized_content
def queue_exists(transport, request, name, callback=None):
"""Checks if the queue exists."""
try:
_common_queue_ops('queue_exists', transport,
request, name, callback=callback)
return True
except errors.ResourceNotFound:
return False
def queue_get(transport, request, name, callback=None):
"""Retrieve a queue."""
return _common_queue_ops('queue_get', transport,
request, name, callback=callback)
def queue_get_metadata(transport, request, name, callback=None):
"""Gets queue metadata."""
return _common_queue_ops('queue_get_metadata', transport,
request, name, callback=callback)
def queue_set_metadata(transport, request, name, metadata, callback=None):
"""Sets queue metadata."""
request.operation = 'queue_set_metadata'
request.params['queue_name'] = name
request.content = json.dumps(metadata)
transport.send(request)
def queue_get_stats(transport, request, name):
return _common_queue_ops('queue_get_stats', transport,
request, name)
def queue_delete(transport, request, name, callback=None):
"""Deletes queue."""
return _common_queue_ops('queue_delete', transport,
request, name, callback=callback)
def queue_list(transport, request, callback=None, **kwargs):
"""Gets a list of queues
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
:param kwargs: Optional arguments for this operation.
- marker: Where to start getting queues from.
- limit: Maximum number of queues to get.
"""
request.operation = 'queue_list'
request.params.update(kwargs)
resp = transport.send(request)
if not resp.content:
return {'links': [], 'queues': []}
return resp.deserialized_content
def queue_purge(transport, request, name, resource_types=None):
"""Purge resources under a queue
@@ -109,6 +201,468 @@ def queue_purge(transport, request, name, resource_types=None):
return resp.deserialized_content
def message_list(transport, request, queue_name, callback=None, **kwargs):
"""Gets a list of messages in queue `queue_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param queue_name: Queue reference name.
:type queue_name: str
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
:param kwargs: Optional arguments for this operation.
- marker: Where to start getting messages from.
- limit: Maximum number of messages to get.
- echo: Whether to get our own messages.
- include_claimed: Whether to include claimed
messages.
"""
request.operation = 'message_list'
request.params['queue_name'] = queue_name
# NOTE(flaper87): Assume passed params
# are accepted by the API, if not, the
# API itself will raise an error.
request.params.update(kwargs)
resp = transport.send(request)
if not resp.content:
# NOTE(flaper87): We could also return None
# or an empty dict, however, we're giving
# more value to a consistent API here by
# returning a compliant dict with empty
# `links` and `messages`
return {'links': [], 'messages': []}
return resp.deserialized_content
def message_post(transport, request, queue_name, messages, callback=None):
"""Post messages to `queue_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param queue_name: Queue reference name.
:type queue_name: str
:param messages: One or more messages to post.
:param messages: `list`
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = 'message_post'
request.params['queue_name'] = queue_name
request.content = json.dumps(messages)
resp = transport.send(request)
return resp.deserialized_content
def message_get(transport, request, queue_name, message_id, callback=None):
"""Gets one message from the queue by id
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param queue_name: Queue reference name.
:type queue_name: str
:param message_id: Message reference.
:param message_id: str
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = 'message_get'
request.params['queue_name'] = queue_name
request.params['message_id'] = message_id
resp = transport.send(request)
return resp.deserialized_content
def message_get_many(transport, request, queue_name, messages, callback=None):
"""Gets many messages by id
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param queue_name: Queue reference name.
:type queue_name: str
:param messages: Messages references.
:param messages: list of str
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = 'message_get_many'
request.params['queue_name'] = queue_name
request.params['ids'] = messages
resp = transport.send(request)
return resp.deserialized_content
def message_delete(transport, request, queue_name, message_id,
claim_id=None, callback=None):
"""Deletes messages from `queue_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param queue_name: Queue reference name.
:type queue_name: str
:param message_id: Message reference.
:param message_id: str
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = 'message_delete'
request.params['queue_name'] = queue_name
request.params['message_id'] = message_id
if claim_id:
request.params['claim_id'] = claim_id
transport.send(request)
def message_delete_many(transport, request, queue_name,
ids, callback=None):
"""Deletes `ids` messages from `queue_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param queue_name: Queue reference name.
:type queue_name: str
:param ids: Ids of the messages to delete
:type ids: List of str
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = 'message_delete_many'
request.params['queue_name'] = queue_name
request.params['ids'] = ids
transport.send(request)
def message_pop(transport, request, queue_name,
count, callback=None):
"""Pops out `count` messages from `queue_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param queue_name: Queue reference name.
:type queue_name: str
:param count: Number of messages to pop.
:type count: int
:param callback: Optional callable to use as callback.
If specified, this request will be sent asynchronously.
(IGNORED UNTIL ASYNC SUPPORT IS COMPLETE)
:type callback: Callable object.
"""
request.operation = 'message_delete_many'
request.params['queue_name'] = queue_name
request.params['pop'] = count
resp = transport.send(request)
return resp.deserialized_content
def claim_create(transport, request, queue_name, **kwargs):
"""Creates a Claim `claim_id` on the queue `queue_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
"""
request.operation = 'claim_create'
request.params['queue_name'] = queue_name
if 'limit' in kwargs:
request.params['limit'] = kwargs.pop('limit')
request.content = json.dumps(kwargs)
resp = transport.send(request)
return resp.deserialized_content
def claim_get(transport, request, queue_name, claim_id):
"""Gets a Claim `claim_id`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
"""
request.operation = 'claim_get'
request.params['queue_name'] = queue_name
request.params['claim_id'] = claim_id
resp = transport.send(request)
return resp.deserialized_content
def claim_update(transport, request, queue_name, claim_id, **kwargs):
"""Updates a Claim `claim_id`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
"""
request.operation = 'claim_update'
request.params['queue_name'] = queue_name
request.params['claim_id'] = claim_id
request.content = json.dumps(kwargs)
resp = transport.send(request)
return resp.deserialized_content
def claim_delete(transport, request, queue_name, claim_id):
"""Deletes a Claim `claim_id`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
"""
request.operation = 'claim_delete'
request.params['queue_name'] = queue_name
request.params['claim_id'] = claim_id
transport.send(request)
def pool_get(transport, request, pool_name, callback=None):
"""Gets pool data
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param pool_name: Pool reference name.
:type pool_name: str
"""
request.operation = 'pool_get'
request.params['pool_name'] = pool_name
resp = transport.send(request)
return resp.deserialized_content
def pool_create(transport, request, pool_name, pool_data):
"""Creates a pool called `pool_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param pool_name: Pool reference name.
:type pool_name: str
:param pool_data: Pool's properties, i.e: weight, uri, options.
:type pool_data: `dict`
"""
request.operation = 'pool_create'
request.params['pool_name'] = pool_name
request.content = json.dumps(pool_data)
transport.send(request)
def pool_update(transport, request, pool_name, pool_data):
"""Updates the pool `pool_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param pool_name: Pool reference name.
:type pool_name: str
:param pool_data: Pool's properties, i.e: weight, uri, options.
:type pool_data: `dict`
"""
request.operation = 'pool_update'
request.params['pool_name'] = pool_name
request.content = json.dumps(pool_data)
resp = transport.send(request)
return resp.deserialized_content
def pool_list(transport, request, **kwargs):
"""Gets a list of pools
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param kwargs: Optional arguments for this operation.
- marker: Where to start getting pools from.
- limit: Maximum number of pools to get.
"""
request.operation = 'pool_list'
request.params.update(kwargs)
resp = transport.send(request)
if not resp.content:
return {'links': [], 'pools': []}
return resp.deserialized_content
def pool_delete(transport, request, pool_name):
"""Deletes the pool `pool_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param pool_name: Pool reference name.
:type pool_name: str
"""
request.operation = 'pool_delete'
request.params['pool_name'] = pool_name
transport.send(request)
def flavor_create(transport, request, name, flavor_data):
"""Creates a flavor called `name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param name: Flavor reference name.
:type name: str
:param flavor_data: Flavor's properties, i.e: pool, capabilities.
:type flavor_data: `dict`
"""
request.operation = 'flavor_create'
request.params['flavor_name'] = name
request.content = json.dumps(flavor_data)
transport.send(request)
def flavor_get(transport, request, flavor_name, callback=None):
"""Gets flavor data
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param flavor_name: Flavor reference name.
:type flavor_name: str
"""
request.operation = 'flavor_get'
request.params['flavor_name'] = flavor_name
resp = transport.send(request)
return resp.deserialized_content
def flavor_update(transport, request, flavor_name, flavor_data):
"""Updates the flavor `flavor_name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param flavor_name: Flavor reference name.
:type flavor_name: str
:param flavor_data: Flavor's properties, i.e: pool, capabilities.
:type flavor_data: `dict`
"""
request.operation = 'flavor_update'
request.params['flavor_name'] = flavor_name
request.content = json.dumps(flavor_data)
resp = transport.send(request)
return resp.deserialized_content
def flavor_list(transport, request, **kwargs):
"""Gets a list of flavors
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param kwargs: Optional arguments for this operation.
- marker: Where to start getting flavors from.
- limit: Maximum number of flavors to get.
"""
request.operation = 'flavor_list'
request.params.update(kwargs)
resp = transport.send(request)
if not resp.content:
return {'links': [], 'flavors': []}
return resp.deserialized_content
def flavor_delete(transport, request, name):
"""Deletes the flavor `name`
:param transport: Transport instance to use
:type transport: `transport.base.Transport`
:param request: Request instance ready to be sent.
:type request: `transport.request.Request`
:param name: Flavor reference name.
:type name: str
"""
request.operation = 'flavor_delete'
request.params['flavor_name'] = name
transport.send(request)
def signed_url_create(transport, request, queue_name, paths=None,
ttl_seconds=None, project_id=None, methods=None):
"""Creates a signed URL given a queue name

58
zaqarclient/queues/v2/flavor.py
View File

@@ -1,4 +1,4 @@
# Copyright (c) 2015 Catalyst IT Ltd.
# Copyright (c) 2014 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.
@@ -13,8 +13,58 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from zaqarclient.queues.v1 import flavor
from zaqarclient.queues.v2 import core
class Flavor(flavor.Flavor):
pass
class Flavor(object):
def __init__(self, client, name,
pool_list=None, auto_create=True,
**kwargs):
self.client = client
self.name = name
self.pool_list = pool_list
self.capabilities = kwargs.get('capabilities', {})
if auto_create:
self.ensure_exists()
def ensure_exists(self):
"""Ensures pool exists
This method is not race safe,
the pool could've been deleted
right after it was called.
"""
req, trans = self.client._request_and_transport()
# As of now on PUT, zaqar server updates flavor if it is already
# exists else it will create a new one. The zaqar client should
# maitain symmetry with zaqar server.
# TBD(mdnadeem): Have to change this code when zaqar server
# behaviour change for PUT operation.
data = {'pool_list': self.pool_list}
req, trans = self.client._request_and_transport()
core.flavor_create(trans, req, self.name, data)
def update(self, flavor_data):
req, trans = self.client._request_and_transport()
core.flavor_update(trans, req, self.name, flavor_data)
for key, value in flavor_data.items():
setattr(self, key, value)
def delete(self):
req, trans = self.client._request_and_transport()
core.flavor_delete(trans, req, self.name)
def get(self):
req, trans = self.client._request_and_transport()
return core.flavor_get(trans, req, self.name, callback=None)
def create_object(parent):
return lambda kwargs: Flavor(parent, kwargs.pop('name'),
auto_create=False, **kwargs)

0
zaqarclient/queues/v1/iterator.py → zaqarclient/queues/v2/iterator.py
View File

8
zaqarclient/queues/v2/message.py
View File

@@ -12,12 +12,16 @@
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Implements a message controller that understands Zaqar messages."""
from zaqarclient.queues.v1 import message
from zaqarclient.queues.v2 import core
class Message(message.Message):
class Message(object):
"""A handler for Zaqar server Message resources.
Attributes are only downloaded once - at creation time.
"""
def __init__(self, queue, ttl, age, body, href=None, id=None,
claim_id=None, claim_count=0, checksum=None):
self.queue = queue

65
zaqarclient/queues/v2/pool.py
View File

@@ -1,4 +1,4 @@
# Copyright (c) 2015 Catalyst IT Ltd.
# Copyright (c) 2014 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.
@@ -13,8 +13,65 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from zaqarclient.queues.v1 import pool
from zaqarclient.queues.v2 import core
class Pool(pool.Pool):
pass
class Pool(object):
def __init__(self, client, name,
weight=None, uri=None,
flavor=None, auto_create=True,
**kwargs):
self.client = client
self.uri = uri
self.name = name
self.weight = weight
self.flavor = flavor
self.options = kwargs.get("options", {})
if auto_create:
self.ensure_exists()
def ensure_exists(self):
"""Ensures pool exists
This method is not race safe,
the pool could've been deleted
right after it was called.
"""
req, trans = self.client._request_and_transport()
# As of now on PUT, zaqar server updates pool if it is already
# exists else it will create a new one. The zaqar client should
# maitain symmetry with zaqar server.
# TBD(mdnadeem): Have to change this code when zaqar server
# behaviour change for PUT operation.
data = {'uri': self.uri,
'weight': self.weight,
'options': self.options}
if self.flavor:
data['flavor'] = self.flavor
req, trans = self.client._request_and_transport()
core.pool_create(trans, req, self.name, data)
def update(self, pool_data):
req, trans = self.client._request_and_transport()
core.pool_update(trans, req, self.name, pool_data)
for key, value in pool_data.items():
setattr(self, key, value)
def delete(self):
req, trans = self.client._request_and_transport()
core.pool_delete(trans, req, self.name)
def get(self):
req, trans = self.client._request_and_transport()
return core.pool_get(trans, req, self.name, callback=None)
def create_object(parent):
return lambda kwargs: Pool(parent, kwargs.pop('name'),
auto_create=False, **kwargs)

210
zaqarclient/queues/v2/queues.py
View File

@@ -13,26 +13,88 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from zaqarclient.queues.v1 import queues
import re
from zaqarclient._i18n import _ # noqa
from zaqarclient.queues.v2 import claim as claim_api
from zaqarclient.queues.v2 import core
from zaqarclient.queues.v2 import iterator
from zaqarclient.queues.v2 import message
# NOTE(wanghao): This is copied from Zaqar server side, so if server have
# updated it someday, we should update it here to keep consistent.
QUEUE_NAME_REGEX = re.compile(r'^[a-zA-Z0-9_\-]+$')
class Queue(queues.Queue):
class Queue(object):
message_module = message
def signed_url(self, paths=None, ttl_seconds=None, methods=None):
req, trans = self.client._request_and_transport()
return core.signed_url_create(trans, req, self._name, paths=paths,
ttl_seconds=ttl_seconds, methods=methods)
def __init__(self, client, name, href=None, metadata=None,
auto_create=True, force_create=False):
"""Initialize queue object
def subscriptions(self, detailed=False, marker=None, limit=20):
return self.client.subscriptions(queue_name=self._name,
detailed=detailed,
marker=marker,
limit=limit)
:param client: The client object of Zaqar.
:type client: `object`
:param name: Name of the queue.
:type name: str
:param href : Hypertext Reference.
:type href: str
:param metadata : A metadata object of the queue.
:type metadata: `dict`
:param auto_create: If create the queue automatically in database.
:type auto_create: `boolean`
:param force_create: If create the queue and skip the API version
check, which is useful for command line interface.
:type force_create: `boolean`
:returns: The queue object.
"""
self.client = client
if name == "":
raise ValueError(_('Queue name does not have a value'))
if not QUEUE_NAME_REGEX.match(str(name)):
raise ValueError(_('The queue name may only contain ASCII '
'letters, digits, underscores and dashes.'))
# NOTE(flaper87) Queue Info
self._name = name
self._metadata = metadata
self._href = href
# NOTE(flwang): If force_create is True, then even though auto_create
# is not True, the queue should be created anyway.
if auto_create or force_create:
self.ensure_exists(force_create=force_create)
@property
def name(self):
return self._name
@property
def href(self):
return self._href
@property
def metadata_dict(self):
return dict(self.metadata())
def exists(self):
"""Checks if the queue exists."""
req, trans = self.client._request_and_transport()
return core.queue_exists(trans, req, self._name)
def ensure_exists(self, force_create=False):
"""Ensures a queue exists
This method is not race safe,
the queue could've been deleted
right after it was called.
"""
req, trans = self.client._request_and_transport()
if force_create:
core.queue_create(trans, req, self._name)
def metadata(self, new_meta=None, force_reload=False):
"""Get metadata and return it
@@ -84,6 +146,132 @@ class Queue(queues.Queue):
return self._metadata
@property
def stats(self):
req, trans = self.client._request_and_transport()
return core.queue_get_stats(trans, req, self._name)
def delete(self):
req, trans = self.client._request_and_transport()
core.queue_delete(trans, req, self._name)
# Messages API
def post(self, messages):
"""Posts one or more messages to this queue
:param messages: One or more messages to post
:type messages: `list` or `dict`
:returns: A dict with the result of this operation.
:rtype: `dict`
"""
if not isinstance(messages, list):
messages = [messages]
messages = {'messages': messages}
req, trans = self.client._request_and_transport()
# TODO(flaper87): Return a list of messages
return core.message_post(trans, req,
self._name, messages)
def message(self, message_id):
"""Gets a message by id
:param message_id: Message's reference
:type message_id: str
:returns: A message
:rtype: `dict`
"""
req, trans = self.client._request_and_transport()
msg = core.message_get(trans, req, self._name,
message_id)
return self.message_module.Message(self, **msg)
def messages(self, *messages, **params):
"""Gets a list of messages from the server
This method returns a list of messages, it can be
used to retrieve a set of messages by id or to
walk through the active messages by using the
collection endpoint.
The `messages` and `params` params are mutually exclusive
and the former has the priority.
:param messages: List of messages' ids to retrieve.
:type messages: *args of str
:param params: Filters to use for getting messages
:type params: **kwargs dict.
:returns: List of messages
:rtype: `list`
"""
req, trans = self.client._request_and_transport()
# TODO(flaper87): Return a MessageIterator.
# This iterator should handle limits, pagination
# and messages deserialization.
if messages:
msgs = core.message_get_many(trans, req,
self._name, messages)
else:
# NOTE(flaper87): It's safe to access messages
# directly. If something wrong happens, the core
# API will raise the right exceptions.
msgs = core.message_list(trans, req,
self._name,
**params)
return iterator._Iterator(self.client,
msgs,
'messages',
self.message_module.create_object(self))
def delete_messages(self, *messages):
"""Deletes a set of messages from the server
:param messages: List of messages' ids to delete.
:type messages: *args of str
"""
req, trans = self.client._request_and_transport()
return core.message_delete_many(trans, req, self._name,
set(messages))
def pop(self, count=1):
"""Pop `count` messages from the server
:param count: Number of messages to pop.
:type count: int
:returns: List of messages
:rtype: `list`
"""
req, trans = self.client._request_and_transport()
msgs = core.message_pop(trans, req, self._name, count=count)
return iterator._Iterator(self.client,
msgs,
'messages',
self.message_module.create_object(self))
def signed_url(self, paths=None, ttl_seconds=None, methods=None):
req, trans = self.client._request_and_transport()
return core.signed_url_create(trans, req, self._name, paths=paths,
ttl_seconds=ttl_seconds, methods=methods)
def subscriptions(self, detailed=False, marker=None, limit=20):
return self.client.subscriptions(queue_name=self._name,
detailed=detailed,
marker=marker,
limit=limit)
def purge(self, resource_types=None):
req, trans = self.client._request_and_transport()
core.queue_purge(trans, req, self._name,

2
zaqarclient/tests/queues/base.py
View File

@@ -33,7 +33,7 @@ class QueuesTestBase(base.TestBase):
transport_cls = dummy.DummyTransport
url = ZAQAR_ENDPOINT
version = 1
version = 2
def setUp(self):
super(QueuesTestBase, self).setUp()

50
zaqarclient/tests/queues/claims.py
View File

@@ -17,13 +17,13 @@ import json
import time
from unittest import mock
from zaqarclient.queues.v1 import claim
from zaqarclient.queues.v2 import claim
from zaqarclient.tests.queues import base
from zaqarclient.transport import errors
from zaqarclient.transport import response
class QueueV1ClaimUnitTest(base.QueuesTestBase):
class QueueV2ClaimUnitTest(base.QueuesTestBase):
def test_claim(self):
result = [{
@@ -41,7 +41,7 @@ class QueueV1ClaimUnitTest(base.QueuesTestBase):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, json.dumps(result))
resp = response.Response(None, json.dumps({'messages': result}))
send_method.return_value = resp
claimed = self.queue.claim(ttl=60, grace=60)
@@ -119,7 +119,7 @@ class QueueV1ClaimUnitTest(base.QueuesTestBase):
# doesn't crash.
class QueuesV1ClaimFunctionalTest(base.QueuesTestBase):
class QueuesV2ClaimFunctionalTest(base.QueuesTestBase):
def test_message_claim_functional(self):
queue = self.client.queue("test_queue")
@@ -164,45 +164,3 @@ class QueuesV1ClaimFunctionalTest(base.QueuesTestBase):
self.assertGreaterEqual(res.age, 0)
time.sleep(2)
self.assertGreaterEqual(res.age, 2)
class QueueV1_1ClaimUnitTest(QueueV1ClaimUnitTest):
def test_claim(self):
result = [{
'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({'messages': result}))
send_method.return_value = resp
claimed = self.queue.claim(ttl=60, grace=60)
# messages doesn't support len()
num_tested = 0
for num, msg in enumerate(claimed):
num_tested += 1
self.assertEqual(result[num]['href'], msg.href)
self.assertEqual(len(result), num_tested)
class QueuesV1_1ClaimFunctionalTest(QueuesV1ClaimFunctionalTest):
pass
class QueueV2ClaimUnitTest(QueueV1_1ClaimUnitTest):
pass
class QueuesV2ClaimFunctionalTest(QueuesV1_1ClaimFunctionalTest):
pass

14
zaqarclient/tests/queues/flavor.py
View File

@@ -16,12 +16,12 @@
import json
from unittest import mock
from zaqarclient.queues.v1 import iterator
from zaqarclient.queues.v2 import iterator
from zaqarclient.tests.queues import base
from zaqarclient.transport import response
class QueuesV1_1FlavorUnitTest(base.QueuesTestBase):
class QueuesV2FlavorUnitTest(base.QueuesTestBase):
def test_flavor_create(self):
pool_list = ['pool1', 'pool2']
@@ -112,7 +112,7 @@ class QueuesV1_1FlavorUnitTest(base.QueuesTestBase):
# doesn't crash.
class QueuesV1_1FlavorFunctionalTest(base.QueuesTestBase):
class QueuesV2FlavorFunctionalTest(base.QueuesTestBase):
def test_flavor_create(self):
pool_data = {'uri': 'mongodb://127.0.0.1:27017',
@@ -184,11 +184,3 @@ class QueuesV1_1FlavorFunctionalTest(base.QueuesTestBase):
flavor_data = {'pool_list': pool_list}
flavor = self.client.flavor('tasty', **flavor_data)
flavor.delete()
class QueuesV2FlavorUnitTest(QueuesV1_1FlavorUnitTest):
pass
class QueuesV2FlavorFunctionalTest(QueuesV1_1FlavorFunctionalTest):
pass

14
zaqarclient/tests/queues/pool.py
View File

@@ -16,12 +16,12 @@
import json
from unittest import mock
from zaqarclient.queues.v1 import iterator
from zaqarclient.queues.v2 import iterator
from zaqarclient.tests.queues import base
from zaqarclient.transport import response
class QueuesV1PoolUnitTest(base.QueuesTestBase):
class QueuesV2PoolUnitTest(base.QueuesTestBase):
def test_pool_create(self):
pool_data = {'weight': 10,
@@ -125,7 +125,7 @@ class QueuesV1PoolUnitTest(base.QueuesTestBase):
# doesn't crash.
class QueuesV1_1PoolFunctionalTest(base.QueuesTestBase):
class QueuesV2PoolFunctionalTest(base.QueuesTestBase):
def test_pool_get(self):
pool_data = {'weight': 10,
@@ -173,11 +173,3 @@ class QueuesV1_1PoolFunctionalTest(base.QueuesTestBase):
pool = self.client.pool('FuncTestPool', **pool_data)
pool.delete()
class QueuesV2PoolUnitTest(QueuesV1PoolUnitTest):
pass
class QueuesV2PoolFunctionalTest(QueuesV1_1PoolFunctionalTest):
pass

371
zaqarclient/tests/queues/queues.py
View File

@@ -18,42 +18,49 @@ from unittest 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 iterator
from zaqarclient.queues.v2 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)
class QueuesV2QueueUnitTest(base.QueuesTestBase):
def test_queue_metadata_update(self):
test_metadata = {'type': 'Bank Accounts'}
new_meta = {'flavor': 'test'}
test_metadata = {'type': 'Bank Accounts', 'name': 'test1'}
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)
# add 'test_metadata'
metadata = self.queue.metadata(new_meta=test_metadata)
self.assertEqual(test_metadata, metadata)
metadata = self.queue.metadata(new_meta)
self.assertEqual(new_meta, metadata)
new_metadata_replace = {'type': 'test', 'name': 'test1'}
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, json.dumps(new_metadata_replace))
send_method.return_value = resp
# repalce 'type'
metadata = self.queue.metadata(
new_meta=new_metadata_replace)
expect_metadata = {'type': 'test', "name": 'test1'}
self.assertEqual(expect_metadata, metadata)
remove_metadata = {'name': 'test1'}
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, json.dumps(remove_metadata))
send_method.return_value = resp
# remove 'type'
metadata = self.queue.metadata(new_meta=remove_metadata)
expect_metadata = {"name": 'test1'}
self.assertEqual(expect_metadata, metadata)
def test_queue_create(self):
with mock.patch.object(self.transport, 'send',
@@ -178,27 +185,6 @@ class QueuesV1QueueUnitTest(base.QueuesTestBase):
# 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',
@@ -241,27 +227,106 @@ class QueuesV1QueueUnitTest(base.QueuesTestBase):
# just checking our way down to the transport
# doesn't crash.
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'}
}]
class QueuesV1QueueFunctionalTest(base.QueuesTestBase):
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
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())
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.
def test_queue_subscriptions(self):
result = {
"subscriptions": [{
"source": 'test',
"id": "1",
"subscriber": 'http://trigger.me',
"ttl": 3600,
"age": 1800,
"confirmed": False,
"options": {}},
{
"source": 'test',
"id": "2",
"subscriber": 'http://trigger.you',
"ttl": 7200,
"age": 1800,
"confirmed": False,
"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)
def test_queue_purge(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.purge()
# NOTE(flwang): Nothing to assert here,
# just checking our way down to the transport
# doesn't crash.
def test_queue_purge_messages(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.purge(resource_types=['messages'])
self.assertEqual({"resource_types": ["messages"]},
json.loads(send_method.call_args[0][0].content))
class QueuesV2QueueFunctionalTest(base.QueuesTestBase):
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())
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.assertFalse(queue.exists())
self.assertEqual(0, len(list(queue.messages(echo=True))))
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())
queue = self.client.queue("404")
self.assertRaises(errors.InvalidOperation, queue.exists)
def test_queue_stats_functional(self):
messages = [
@@ -277,27 +342,6 @@ class QueuesV1QueueFunctionalTest(base.QueuesTestBase):
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['type'], metadata['type'])
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['type'], metadata['type'])
def test_message_post_functional(self):
messages = [
{'ttl': 60, 'body': 'Post It!'},
@@ -399,72 +443,6 @@ class QueuesV1QueueFunctionalTest(base.QueuesTestBase):
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.
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
self.assertRaises(RuntimeError, self.queue.metadata, test_metadata)
def test_queue_metadata_update(self):
# v1.1 doesn't support set queue metadata
pass
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)
@@ -484,117 +462,6 @@ class QueuesV1_1QueueFunctionalTest(QueuesV1QueueFunctionalTest):
remaining = queue.messages(echo=True)
self.assertEqual(1, len(list(remaining)))
def test_queue_metadata_functional(self):
# v1.1 doesn't support set queue metadata
pass
def test_queue_metadata_reload_functional(self):
# v1.1 doesn't support set queue metadata
pass
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,
"age": 1800,
"confirmed": False,
"options": {}},
{
"source": 'test',
"id": "2",
"subscriber": 'http://trigger.you',
"ttl": 7200,
"age": 1800,
"confirmed": False,
"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)
def test_queue_metadata(self):
# checked in "test_queue_metadata_update"
pass
def test_queue_metadata_update(self):
test_metadata = {'type': 'Bank Accounts', 'name': 'test1'}
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
# add 'test_metadata'
metadata = self.queue.metadata(new_meta=test_metadata)
self.assertEqual(test_metadata, metadata)
new_metadata_replace = {'type': 'test', 'name': 'test1'}
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, json.dumps(new_metadata_replace))
send_method.return_value = resp
# repalce 'type'
metadata = self.queue.metadata(
new_meta=new_metadata_replace)
expect_metadata = {'type': 'test', "name": 'test1'}
self.assertEqual(expect_metadata, metadata)
remove_metadata = {'name': 'test1'}
with mock.patch.object(self.transport, 'send',
autospec=True) as send_method:
resp = response.Response(None, json.dumps(remove_metadata))
send_method.return_value = resp
# remove 'type'
metadata = self.queue.metadata(new_meta=remove_metadata)
expect_metadata = {"name": 'test1'}
self.assertEqual(expect_metadata, metadata)
def test_queue_purge(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.purge()
# NOTE(flwang): Nothing to assert here,
# just checking our way down to the transport
# doesn't crash.
def test_queue_purge_messages(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.purge(resource_types=['messages'])
self.assertEqual({"resource_types": ["messages"]},
json.loads(send_method.call_args[0][0].content))
class QueuesV2QueueFunctionalTest(QueuesV1_1QueueFunctionalTest):
def test_signed_url(self):
queue = self.client.queue('test_queue')
messages = [{'ttl': 300, 'body': 'Post It!'}]

4
zaqarclient/transport/http.py
View File

@@ -16,7 +16,6 @@
import json
from oslo_utils import importutils
from oslo_utils import versionutils
from zaqarclient.common import http
from zaqarclient.transport import base
@@ -78,8 +77,7 @@ class HttpTransport(base.Transport):
# NOTE(flape87): Do not modify
# request's headers directly.
headers = request.headers.copy()
if (request.operation == 'queue_update' and
versionutils.is_compatible(request.api.label, 'v2')):
if request.operation == 'queue_update':
headers['content-type'] = \
'application/openstack-messaging-v2.0-json-patch'
else:

2
zaqarclient/transport/request.py
View File

@@ -85,7 +85,7 @@ class Request(object):
:type params: dict
:param headers: Request headers. Default: None
:type headers: dict
:param api: Api entry point. i.e: 'queues.v1'
:param api: Api entry point. i.e: 'queues.v2'
:type api: str.
:param verify: If verify the SSL cert
:type verify: bool