Implement queue's API methods
This patch implements methods for every queue operation. The 2
operations left out in this patch are: queue_list and queue_stats.
This patch makes the client, officially, usable as a library. An example
of how this could be use:
client = Client()
queue = client.queue('my_queue') # Creates the queue automatically
queue.metadata() # Get's metadata from the queue
queue.metadata(dict(new_meta='Yo yo!'))
queue.delete()
Metadata was implemented as method instead of as a property to make it
explicit that both get and set do something in the remote server.
The client API was implemented in 2 separate levels. The highest level
is the one shown in the example above, which allows users to just query
Marconi instances without worrying about the transport, request build
process, references, deserealization and what not. The lowest instead,
allows the user to control every bit of the communication process - the
user can pick a specific transport, or pass custom params. For example -
which will make it possible to have other implementations around that
API in addition to the asynchronous support. Example of the API:
transport = http.HttpTransport(cfg.CONF)
request = request.prepare_request(cfg.CONF,
endpoint='http://localhost:8888')
core.queue_create(transport, request, 1, callback=my_callback)
There are tons of things to do and improve. For example:
* The way config params are registered may work when using
marconiclient as a library but they won't when using it as a CLI
tool.
* This code lacks of logging.
* Handling of 20(1|4) is missing.
Partially-Implements blueprint python-marconiclient-v1
Implements blueprint queues-management
Change-Id: I8bdc8a4aff8ea22b5673bc7440e07796ecaf34cc
This commit is contained in:
Flavio Percoco
parent
0dd1dff5c6
commit
6d0391ceb3
|
|
@ -12,7 +12,7 @@
|
|||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
__all__ = ['MarconiError', 'InvalidOperation']
|
||||
__all__ = ['MarconiError', 'DriverLoadFailure', 'InvalidOperation']
|
||||
|
||||
|
||||
class MarconiError(Exception):
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
# Copyright (c) 2013 Rackspace, Inc.
|
||||
# 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.
|
||||
|
|
@ -13,55 +13,71 @@
|
|||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import collections
|
||||
from marconiclient.transport import api
|
||||
|
||||
|
||||
ApiInfo = collections.namedtuple('ApiInfo', 'mandatory optional')
|
||||
class V1(api.Api):
|
||||
|
||||
_API_DATA = dict(
|
||||
create_queue=ApiInfo(
|
||||
mandatory=set(['queue_name']), optional=set()),
|
||||
list_queues=ApiInfo(
|
||||
mandatory=set(), optional=set(['marker', 'limit', 'detailed'])),
|
||||
queue_exists=ApiInfo(mandatory=set(['queue_name']), optional=set()),
|
||||
delete_queue=ApiInfo(mandatory=set(['queue_name']), optional=set()),
|
||||
set_queue_metadata=ApiInfo(
|
||||
mandatory=set(['queue_name', 'metadata']), optional=set()),
|
||||
get_queue_metadata=ApiInfo(
|
||||
mandatory=set(['queue_name']), optional=set()),
|
||||
get_queue_stats=ApiInfo(mandatory=set(['queue_name']), optional=set()),
|
||||
list_messages=ApiInfo(
|
||||
mandatory=set(['queue_name']),
|
||||
optional=set(['marker', 'limit', 'echo', 'include_claimed'])),
|
||||
get_message=ApiInfo(
|
||||
mandatory=set(['queue_name', 'message_id']),
|
||||
optional=set(['claim_id'])),
|
||||
get_messages_by_id=ApiInfo(
|
||||
mandatory=set(['queue_name', 'message_ids']),
|
||||
optional=set()),
|
||||
post_messages=ApiInfo(
|
||||
mandatory=set(['queue_name', 'messagedata']), optional=set()),
|
||||
delete_message=ApiInfo(
|
||||
mandatory=set(['queue_name', 'message_id']),
|
||||
optional=set(['claim_id'])),
|
||||
delete_messages_by_id=ApiInfo(
|
||||
mandatory=set(['queue_name', 'message_ids']), optional=set()),
|
||||
claim_messages=ApiInfo(
|
||||
mandatory=set(['queue_name', 'ttl', 'grace_period']),
|
||||
optional=set(['limit'])),
|
||||
query_claim=ApiInfo(
|
||||
mandatory=set(['queue_name', 'claim_id']), optional=set()),
|
||||
update_claim=ApiInfo(
|
||||
mandatory=set(['queue_name', 'claim_id', 'ttl']), optional=set()),
|
||||
release_claim=ApiInfo(
|
||||
mandatory=set(['queue_name', 'claim_id']), optional=set()),
|
||||
)
|
||||
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'}
|
||||
},
|
||||
},
|
||||
|
||||
def info():
|
||||
"""A dict where the keys and values are valid operations and `ApiInfo`
|
||||
named tuples respectively.
|
||||
The `ApiInfo` named tuples have a `mandatory` and an `optional` property
|
||||
that list the params for the respective operation.
|
||||
"""
|
||||
return _API_DATA.copy()
|
||||
'queue_exists': {
|
||||
'ref': 'queues/{queue_name}',
|
||||
'method': 'HEAD',
|
||||
'properties': {
|
||||
'queue_name': {'type': 'string'}
|
||||
}
|
||||
},
|
||||
|
||||
'queue_delete': {
|
||||
'ref': 'queues/{queue_name}',
|
||||
'method': 'DELETE',
|
||||
'properties': {
|
||||
'queue_name': {'type': 'string'}
|
||||
}
|
||||
},
|
||||
|
||||
'queue_set_metadata': {
|
||||
'ref': 'queues/{queue_name}/metadata',
|
||||
'method': 'PUT',
|
||||
'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',
|
||||
'properties': {
|
||||
'queue_name': {'type': 'string'}
|
||||
}
|
||||
},
|
||||
|
||||
'queue_get_stats': {
|
||||
'ref': 'queues/{queue_name}/stats',
|
||||
'method': 'GET',
|
||||
'properties': {
|
||||
'queue_name': {'type': 'string'}
|
||||
}
|
||||
},
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,54 @@
|
|||
# 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.config import cfg
|
||||
|
||||
from marconiclient.queues.v1 import queues
|
||||
from marconiclient import transport
|
||||
|
||||
|
||||
_CLIENT_OPTIONS = [
|
||||
cfg.StrOpt('os_queues_url',
|
||||
help='Queues remote URL'),
|
||||
]
|
||||
|
||||
|
||||
class Client(object):
|
||||
|
||||
def __init__(self, conf, url=None, version=1):
|
||||
self.conf = conf
|
||||
|
||||
# NOTE(flaper87): This won't actually register
|
||||
# the CLI options until the class is instantiated
|
||||
# which is dumb. It'll refactored when the CLI API
|
||||
# work starts.
|
||||
self.conf.register_cli_opts(_CLIENT_OPTIONS)
|
||||
self.api_url = self.conf.os_queues_url or url
|
||||
self.api_version = version
|
||||
|
||||
def transport(self):
|
||||
"""Gets a transport based on conf."""
|
||||
return transport.get_transport_for_conf(self.conf)
|
||||
|
||||
def queue(self, ref, **kwargs):
|
||||
"""Returns a queue instance
|
||||
|
||||
:param ref: Queue's reference id.
|
||||
:type ref: `six.text_type`
|
||||
|
||||
:returns: A queue instance
|
||||
:rtype: `queues.Queue`
|
||||
"""
|
||||
return queues.Queue(self, ref, **kwargs)
|
||||
|
|
@ -0,0 +1,93 @@
|
|||
# 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 marconiclient.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: `six.text_type`
|
||||
: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
|
||||
return transport.send(request)
|
||||
|
||||
|
||||
def queue_create(transport, request, name, callback=None):
|
||||
"""Creates a queue."""
|
||||
return _common_queue_ops('queue_create', transport,
|
||||
request, name, callback=callback)
|
||||
|
||||
|
||||
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_metadata(transport, request, name, callback=None):
|
||||
"""Gets queue metadata."""
|
||||
resp = _common_queue_ops('queue_get_metadata', transport,
|
||||
request, name, callback=callback)
|
||||
return json.loads(resp.content)
|
||||
|
||||
|
||||
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_delete(transport, request, name, callback=None):
|
||||
"""Deletes queue."""
|
||||
return _common_queue_ops('queue_delete', transport,
|
||||
request, name, callback=callback)
|
||||
|
|
@ -0,0 +1,103 @@
|
|||
# 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 marconiclient.queues.v1 import core
|
||||
from marconiclient import transport
|
||||
from marconiclient.transport import request
|
||||
|
||||
|
||||
class Queue(object):
|
||||
|
||||
def __init__(self, client, queue_id, auto_create=True):
|
||||
self.client = client
|
||||
|
||||
# NOTE(flaper87) Queue Info
|
||||
self._id = queue_id
|
||||
self._metadata = None
|
||||
|
||||
if auto_create:
|
||||
self.ensure_exists()
|
||||
|
||||
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: `transport.request.Request`
|
||||
"""
|
||||
|
||||
trans = transport.get_transport_for(self.client.conf, request)
|
||||
return (trans or self.client.transport)
|
||||
|
||||
def _request_and_transport(self):
|
||||
api = 'queues.v' + str(self.client.api_version)
|
||||
req = request.prepare_request(self.client.conf,
|
||||
endpoint=self.client.api_url,
|
||||
api=api)
|
||||
|
||||
trans = self._get_transport(req)
|
||||
return req, trans
|
||||
|
||||
def exists(self):
|
||||
"""Checks if the queue exists."""
|
||||
req, trans = self._request_and_transport()
|
||||
return core.queue_exists(trans, req, self._id)
|
||||
|
||||
def ensure_exists(self):
|
||||
"""Ensures a queue exists
|
||||
|
||||
This method is not race safe,
|
||||
the queue could've been deleted
|
||||
right after it was called.
|
||||
"""
|
||||
req, trans = self._request_and_transport()
|
||||
core.queue_create(trans, req, self._id)
|
||||
|
||||
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.
|
||||
: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._request_and_transport()
|
||||
|
||||
if new_meta:
|
||||
core.queue_set_metadata(trans, req, self._id, new_meta)
|
||||
self._metadata = new_meta
|
||||
|
||||
# TODO(flaper87): Cache with timeout
|
||||
if self._metadata and not force_reload:
|
||||
return self._metadata
|
||||
|
||||
self._metadata = core.queue_get_metadata(trans, req, self._id)
|
||||
return self._metadata
|
||||
|
||||
def delete(self):
|
||||
req, trans = self._request_and_transport()
|
||||
core.queue_delete(trans, req, self._id)
|
||||
|
|
@ -0,0 +1,156 @@
|
|||
# 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 os
|
||||
|
||||
import json
|
||||
import mock
|
||||
import testtools
|
||||
|
||||
from marconiclient.queues.v1 import client
|
||||
from marconiclient.tests import base
|
||||
from marconiclient.transport import response
|
||||
|
||||
_RUN_FUNCTIONAL = os.environ.get('MARCONICLIENT_TEST_FUNCTIONAL', False)
|
||||
|
||||
|
||||
class QueuesV1QueueTestBase(base.TestBase):
|
||||
|
||||
transport_cls = None
|
||||
|
||||
# NOTE(flaper87): These class attributes
|
||||
# are intended for functional tests only
|
||||
# and will be replaced with something
|
||||
# dynamically loaded to allow tests against
|
||||
# remote instances.
|
||||
url = None
|
||||
version = None
|
||||
|
||||
def setUp(self):
|
||||
super(QueuesV1QueueTestBase, self).setUp()
|
||||
self.transport = self.transport_cls(self.conf)
|
||||
|
||||
self.client = client.Client(self.conf, self.url,
|
||||
self.version)
|
||||
|
||||
# NOTE(flaper87): Nasty monkeypatch, lets use
|
||||
# the dummy transport here.
|
||||
#setattr(self.client, 'transport', self.transport)
|
||||
self.queue = self.client.queue(1, auto_create=False)
|
||||
self.queue._get_transport = mock.Mock(return_value=self.transport)
|
||||
|
||||
self.is_functional = _RUN_FUNCTIONAL
|
||||
|
||||
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()
|
||||
self.assertEqual(metadata, test_metadata)
|
||||
|
||||
def test_queue_create(self):
|
||||
with mock.patch.object(self.transport, 'send',
|
||||
autospec=True) as send_method:
|
||||
|
||||
resp = response.Response(None, None)
|
||||
send_method.return_value = resp
|
||||
|
||||
self.queue.ensure_exists()
|
||||
|
||||
# NOTE(flaper87): Nothing to assert here,
|
||||
# just checking our way down to the transport
|
||||
# doesn't crash.
|
||||
|
||||
def test_queue_delete(self):
|
||||
with mock.patch.object(self.transport, 'send',
|
||||
autospec=True) as send_method:
|
||||
|
||||
resp = response.Response(None, None)
|
||||
send_method.return_value = resp
|
||||
|
||||
self.queue.delete()
|
||||
|
||||
# NOTE(flaper87): Nothing to assert here,
|
||||
# just checking our way down to the transport
|
||||
# doesn't crash.
|
||||
|
||||
def test_queue_exists(self):
|
||||
with mock.patch.object(self.transport, 'send',
|
||||
autospec=True) as send_method:
|
||||
|
||||
resp = response.Response(None, None)
|
||||
send_method.return_value = resp
|
||||
|
||||
self.queue.exists()
|
||||
|
||||
# NOTE(flaper87): Nothing to assert here,
|
||||
# just checking our way down to the transport
|
||||
# doesn't crash.
|
||||
|
||||
|
||||
class QueuesV1QueueFuncMixin(object):
|
||||
|
||||
@testtools.skipUnless(_RUN_FUNCTIONAL,
|
||||
'Functional tests disabled')
|
||||
def test_queue_create_functional(self):
|
||||
queue = self.client.queue("nonono")
|
||||
queue._get_transport = mock.Mock(return_value=self.transport)
|
||||
self.assertTrue(queue.exists())
|
||||
|
||||
@testtools.skipUnless(_RUN_FUNCTIONAL,
|
||||
'Functional tests disabled')
|
||||
def test_queue_delete_functional(self):
|
||||
queue = self.client.queue("nonono")
|
||||
queue._get_transport = mock.Mock(return_value=self.transport)
|
||||
self.assertTrue(queue.exists())
|
||||
queue.delete()
|
||||
self.assertFalse(queue.exists())
|
||||
|
||||
@testtools.skipUnless(_RUN_FUNCTIONAL,
|
||||
'Functional tests disabled')
|
||||
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())
|
||||
|
||||
@testtools.skipUnless(_RUN_FUNCTIONAL,
|
||||
'Functional tests disabled')
|
||||
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(metadata, test_metadata)
|
||||
|
||||
@testtools.skipUnless(_RUN_FUNCTIONAL,
|
||||
'Functional tests disabled')
|
||||
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(metadata, test_metadata)
|
||||
|
|
@ -0,0 +1,23 @@
|
|||
# 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 marconiclient.transport import base
|
||||
|
||||
|
||||
class DummyTransport(base.Transport):
|
||||
|
||||
def send(self, request):
|
||||
pass
|
||||
|
|
@ -13,12 +13,20 @@
|
|||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
from oslo.config import cfg
|
||||
import six
|
||||
from six.moves.urllib import parse
|
||||
from stevedore import driver
|
||||
|
||||
from marconiclient import errors
|
||||
|
||||
_TRANSPORT_OPTIONS = [
|
||||
cfg.StrOpt('default_transport', default='http',
|
||||
help='Transport to use as default'),
|
||||
cfg.IntOpt('default_transport_version', default=1,
|
||||
help='Transport to use as default'),
|
||||
]
|
||||
|
||||
|
||||
def get_transport(conf, transport, version=1):
|
||||
"""Gets a transport and returns it.
|
||||
|
|
@ -48,6 +56,20 @@ def get_transport(conf, transport, version=1):
|
|||
return mgr.driver
|
||||
|
||||
|
||||
def get_transport_for_conf(conf):
|
||||
"""Gets a transport based on the config object
|
||||
|
||||
It'll load a transport based on the `default-transport`
|
||||
and `default-transport-version` params.
|
||||
|
||||
:param conf: the user configuration
|
||||
:type conf: cfg.ConfigOpts
|
||||
"""
|
||||
conf.register_opts(_TRANSPORT_OPTIONS)
|
||||
return get_transport(conf, conf.default_transport,
|
||||
conf.default_transport_version)
|
||||
|
||||
|
||||
def get_transport_for(conf, url_or_request, version=1):
|
||||
"""Gets a transport for a given url.
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,43 @@
|
|||
# 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.
|
||||
|
||||
|
||||
"""
|
||||
Errors below must be used to translate transport specific
|
||||
errors to Marconi errors. For example, HTTP 404s should be
|
||||
raised as `ResourceNotFound`
|
||||
"""
|
||||
|
||||
from marconiclient import errors
|
||||
|
||||
__all__ = ['TransportError', 'ResourceNotFound', 'MalformedRequest']
|
||||
|
||||
|
||||
class TransportError(errors.MarconiError):
|
||||
"""Base class for all transport errors."""
|
||||
|
||||
|
||||
class ResourceNotFound(TransportError):
|
||||
"""Indicates that a resource is missing
|
||||
|
||||
This error maps to HTTP's 404
|
||||
"""
|
||||
|
||||
|
||||
class MalformedRequest(TransportError):
|
||||
"""Indicates that a request is malformed
|
||||
|
||||
This error maps to HTTP's 400
|
||||
"""
|
||||
|
|
@ -13,12 +13,25 @@
|
|||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import json
|
||||
|
||||
from marconiclient.common import http
|
||||
from marconiclient.transport import base
|
||||
# NOTE(flaper87): Something is completely borked
|
||||
# with some imports. Using `from ... import errors`
|
||||
# will end up importing `marconiclient.errors` instead
|
||||
# of transports
|
||||
import marconiclient.transport.errors as errors
|
||||
from marconiclient.transport import response
|
||||
|
||||
|
||||
class HttpTransport(base.Transport):
|
||||
|
||||
http_to_marconi = {
|
||||
400: errors.MalformedRequest,
|
||||
404: errors.ResourceNotFound,
|
||||
}
|
||||
|
||||
def __init__(self, conf):
|
||||
super(HttpTransport, self).__init__(conf)
|
||||
self.client = http.Client()
|
||||
|
|
@ -52,8 +65,23 @@ class HttpTransport(base.Transport):
|
|||
headers = request.headers.copy()
|
||||
headers['content-type'] = 'application/json'
|
||||
|
||||
return self.client.request(method,
|
||||
resp = self.client.request(method,
|
||||
url=url,
|
||||
params=request.params,
|
||||
headers=headers,
|
||||
data=request.content)
|
||||
|
||||
if resp.status_code in self.http_to_marconi:
|
||||
try:
|
||||
msg = json.loads(resp.text)['description']
|
||||
except Exception:
|
||||
# TODO(flaper87): Log this exception
|
||||
# but don't stop raising the corresponding
|
||||
# exception
|
||||
msg = ''
|
||||
raise self.http_to_marconi[resp.status_code](msg)
|
||||
|
||||
# NOTE(flaper87): This reads the whole content
|
||||
# and will consume any attempt of streaming.
|
||||
return response.Response(request, resp.text,
|
||||
headers=resp.headers)
|
||||
|
|
|
|||
|
|
@ -33,6 +33,9 @@ packages =
|
|||
marconiclient.transport =
|
||||
http.v1 = marconiclient.transport.http:HttpTransport
|
||||
|
||||
marconiclient.api =
|
||||
queues.v1 = marconiclient.queues.v1.api:V1
|
||||
|
||||
[nosetests]
|
||||
where=tests
|
||||
verbosity=2
|
||||
|
|
|
|||
|
|
@ -0,0 +1,90 @@
|
|||
# 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 mock
|
||||
|
||||
from marconiclient.queues.v1 import core
|
||||
from marconiclient.tests import base
|
||||
from marconiclient.tests.transport import dummy
|
||||
import marconiclient.transport.errors as errors
|
||||
from marconiclient.transport import request
|
||||
from marconiclient.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 = 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 = 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 = 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 = None
|
||||
|
||||
req = request.Request()
|
||||
core.queue_exists(self.transport, req, update_data, 'test')
|
||||
self.assertIn('queue_name', req.params)
|
||||
|
||||
self.assertIn('queue_name', req.params)
|
||||
|
|
@ -0,0 +1,32 @@
|
|||
# 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 marconiclient.tests.queues import queues
|
||||
from marconiclient.tests.transport import dummy
|
||||
from marconiclient.transport import http
|
||||
|
||||
|
||||
class QueuesV1QueueDummyTransportTest(queues.QueuesV1QueueTestBase):
|
||||
|
||||
transport_cls = dummy.DummyTransport
|
||||
|
||||
|
||||
class QueuesV1QueueHttpTransportTest(queues.QueuesV1QueueTestBase,
|
||||
queues.QueuesV1QueueFuncMixin):
|
||||
|
||||
transport_cls = http.HttpTransport
|
||||
url = 'http://127.0.0.1:8888/v1'
|
||||
version = 1
|
||||
|
|
@ -14,6 +14,7 @@
|
|||
# limitations under the License.
|
||||
|
||||
import mock
|
||||
import requests as prequest
|
||||
|
||||
from marconiclient.tests import base
|
||||
from marconiclient.tests.transport import api
|
||||
|
|
@ -39,7 +40,8 @@ class TestHttpTransport(base.TestBase):
|
|||
with mock.patch.object(self.transport.client, 'request',
|
||||
autospec=True) as request_method:
|
||||
|
||||
request_method.return_value = None
|
||||
resp = prequest.Response()
|
||||
request_method.return_value = resp
|
||||
|
||||
# NOTE(flaper87): Bypass the API
|
||||
# loading step by setting the _api
|
||||
|
|
@ -66,7 +68,8 @@ class TestHttpTransport(base.TestBase):
|
|||
with mock.patch.object(self.transport.client, 'request',
|
||||
autospec=True) as request_method:
|
||||
|
||||
request_method.return_value = None
|
||||
resp = prequest.Response()
|
||||
request_method.return_value = resp
|
||||
self.transport.send(req)
|
||||
|
||||
final_url = 'http://example.org/'
|
||||
|
|
|
|||
Loading…
Reference in New Issue