openstack/zaqar
Code Issues Proposed changes

Merge "Remove support for V1.1 APIs from Zaqar 3"

Browse Source
This commit is contained in:
Zuul
2025-10-11 12:43:19 +00:00
committed by Gerrit Code Review
parent daa2218939 ac800de3e0
commit 34fc0eb123
19 changed files with 1 additions and 3240 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zaqar/tests/functional/wsgi/v1_1/__init__.py
View File
zaqar/tests/functional/wsgi/v1_1/test_claims.py
-276
View File
@@ -1,276 +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.
import uuid
import ddt
from zaqar.tests.functional import base
from zaqar.tests.functional import helpers
@ddt.ddt
class TestClaims(base.V1_1FunctionalTestBase):
"""Tests for Claims."""
server_class = base.ZaqarServer
def setUp(self):
super(TestClaims, self).setUp()
self.headers = helpers.create_zaqar_headers(self.cfg)
self.client.headers = self.headers
self.queue = uuid.uuid1()
self.queue_url = ("{url}/{version}/queues/{queue}".format(
url=self.cfg.zaqar.url,
version="v1.1",
queue=self.queue))
self.client.put(self.queue_url)
self.claim_url = self.queue_url + '/claims'
self.client.set_base_url(self.claim_url)
# Post Messages
url = self.queue_url + '/messages'
doc = helpers.create_message_body_v1_1(
messagecount=self.limits.max_messages_per_page)
for i in range(10):
self.client.post(url, data=doc)
@ddt.data({}, {'limit': 2})
def test_claim_messages(self, params):
"""Claim messages."""
message_count = params.get('limit',
self.limits.max_messages_per_claim_or_pop)
doc = {"ttl": 300, "grace": 100}
result = self.client.post(params=params, data=doc)
self.assertEqual(201, result.status_code)
self.assertSchema(result.json(), 'claim_create')
actual_message_count = len(result.json()['messages'])
self.assertMessageCount(actual_message_count, message_count)
response_headers = set(result.headers.keys())
self.assertIsSubset(self.headers_response_with_body, response_headers)
test_claim_messages.tags = ['smoke', 'positive']
def test_query_claim(self):
"""Query Claim."""
params = {'limit': 1}
doc = {"ttl": 300, "grace": 100}
result = self.client.post(params=params, data=doc)
location = result.headers['Location']
url = self.cfg.zaqar.url + location
result = self.client.get(url)
self.assertEqual(200, result.status_code)
test_query_claim.tags = ['smoke', 'positive']
@ddt.data({}, {"grace": 100})
def test_claim_default_ttl(self, doc):
"""Create claim with default TTL and grace values."""
params = {'limit': 1}
result = self.client.post(params=params, data=doc)
self.assertEqual(201, result.status_code)
location = result.headers['Location']
url = self.cfg.zaqar.url + location
result = self.client.get(url)
self.assertEqual(200, result.status_code)
default_ttl = result.json()['ttl']
self.assertEqual(self.resource_defaults.claim_ttl, default_ttl)
test_claim_default_ttl.tags = ['smoke', 'positive']
def test_claim_more_than_allowed(self):
"""Claim more than max allowed per request.
Zaqar allows a maximum of 20 messages per claim by default.
"""
params = {"limit": self.limits.max_messages_per_claim_or_pop + 1}
doc = {"ttl": 300, "grace": 100}
result = self.client.post(params=params, data=doc)
self.assertEqual(400, result.status_code)
test_claim_more_than_allowed.tags = ['negative']
def test_claim_patch(self):
"""Update Claim."""
# Test Setup - Post Claim
doc = {"ttl": 300, "grace": 400}
result = self.client.post(data=doc)
self.assertEqual(201, result.status_code)
# Patch Claim
claim_location = result.headers['Location']
url = self.cfg.zaqar.url + claim_location
doc_updated = {"ttl": 300, 'grace': 60}
result = self.client.patch(url, data=doc_updated)
self.assertEqual(204, result.status_code)
# verify that the claim TTL is updated
result = self.client.get(url)
new_ttl = result.json()['ttl']
self.assertEqual(doc_updated['ttl'], new_ttl)
test_claim_patch.tags = ['smoke', 'positive']
def test_delete_claimed_message(self):
"""Delete message belonging to a Claim."""
# Test Setup - Post claim
doc = {"ttl": 60, "grace": 60}
result = self.client.post(data=doc)
self.assertEqual(201, result.status_code)
# Delete Claimed Messages
for rst in result.json()['messages']:
href = rst['href']
url = self.cfg.zaqar.url + href
result = self.client.delete(url)
self.assertEqual(204, result.status_code)
test_delete_claimed_message.tags = ['smoke', 'positive']
def test_claim_release(self):
"""Release Claim."""
doc = {"ttl": 300, "grace": 100}
result = self.client.post(data=doc)
self.assertEqual(201, result.status_code)
# Extract claim location and construct the claim URL.
location = result.headers['Location']
url = self.cfg.zaqar.url + location
# Release Claim.
result = self.client.delete(url)
self.assertEqual(204, result.status_code)
test_claim_release.tags = ['smoke', 'positive']
@ddt.data(10000000000000000000, -100, 1, 59, 43201, -10000000000000000000)
def test_claim_invalid_ttl(self, ttl):
"""Post Claim with invalid TTL.
The request JSON body will have a TTL value
outside the allowed range.Allowed ttl values is
60 <= ttl <= 43200.
"""
doc = {"ttl": ttl, "grace": 100}
result = self.client.post(data=doc)
self.assertEqual(400, result.status_code)
test_claim_invalid_ttl.tags = ['negative']
@ddt.data(10000000000000000000, -100, 1, 59, 43201, -10000000000000000000)
def test_claim_invalid_grace(self, grace):
"""Post Claim with invalid grace.
The request JSON body will have a grace value
outside the allowed range.Allowed grace values is
60 <= grace <= 43200.
"""
doc = {"ttl": 100, "grace": grace}
result = self.client.post(data=doc)
self.assertEqual(400, result.status_code)
test_claim_invalid_grace.tags = ['negative']
@ddt.data(0, -100, 30, 10000000000000000000)
def test_claim_invalid_limit(self, grace):
"""Post Claim with invalid limit.
The request url will have a limit outside the allowed range.
Allowed limit values are 0 < limit <= 20(default max).
"""
doc = {"ttl": 100, "grace": grace}
result = self.client.post(data=doc)
self.assertEqual(400, result.status_code)
test_claim_invalid_limit.tags = ['negative']
@ddt.data(10000000000000000000, -100, 1, 59, 43201, -10000000000000000000)
def test_patch_claim_invalid_ttl(self, ttl):
"""Patch Claim with invalid TTL.
The request JSON body will have a TTL value
outside the allowed range.Allowed ttl values is
60 <= ttl <= 43200.
"""
doc = {"ttl": 100, "grace": 100}
result = self.client.post(data=doc)
self.assertEqual(201, result.status_code)
# Extract claim location and construct the claim URL.
location = result.headers['Location']
url = self.cfg.zaqar.url + location
# Patch Claim.
doc = {"ttl": ttl}
result = self.client.patch(url, data=doc)
self.assertEqual(400, result.status_code)
test_patch_claim_invalid_ttl.tags = ['negative']
def test_query_non_existing_claim(self):
"""Query Non Existing Claim."""
path = '/non-existing-claim'
result = self.client.get(path)
self.assertEqual(404, result.status_code)
test_query_non_existing_claim.tags = ['negative']
def test_patch_non_existing_claim(self):
"""Patch Non Existing Claim."""
path = '/non-existing-claim'
doc = {"ttl": 400}
result = self.client.patch(path, data=doc)
self.assertEqual(404, result.status_code)
test_patch_non_existing_claim.tags = ['negative']
def test_delete_non_existing_claim(self):
"""Patch Non Existing Claim."""
path = '/non-existing-claim'
result = self.client.delete(path)
self.assertEqual(204, result.status_code)
test_delete_non_existing_claim.tags = ['negative']
def tearDown(self):
"""Delete Queue after Claim Test."""
super(TestClaims, self).tearDown()
self.client.delete(self.queue_url)
zaqar/tests/functional/wsgi/v1_1/test_health.py
-84
View File
@@ -1,84 +0,0 @@
# Copyright (c) 2014 Catalyst IT Ltd
#
# 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 zaqar.tests.functional import base
from zaqar.tests.functional import helpers
class TestHealth(base.V1_1FunctionalTestBase):
server_class = base.ZaqarAdminServer
config_file = 'wsgi_mongodb_pooled.conf'
def setUp(self):
super(TestHealth, self).setUp()
self.base_url = ("{url}/{version}".format(
url=self.cfg.zaqar.url,
version="v1.1"
))
self.cfg.zaqar.version = "v1.1"
self.headers = helpers.create_zaqar_headers(self.cfg)
self.client.headers = self.headers
self.client.set_base_url(self.base_url)
def test_health_with_pool(self):
# FIXME(flwang): Please use mongodb after the sqlalchemy is disabled
# as pool node and the mongodb is working on gate successfully.
doc = helpers.create_pool_body(
weight=10,
uri=self.mconf['drivers:management_store:mongodb'].uri,
options=dict(database='zaqar_test_pooled_1')
)
pool_name = "pool_1"
result = self.client.put('/pools/' + pool_name, data=doc)
self.assertEqual(201, result.status_code)
queue_name = 'fake_queue'
result = self.client.put('/queues/' + queue_name)
self.assertEqual(201, result.status_code)
sample_messages = {'messages': [
{'body': 239, 'ttl': 999},
{'body': {'key': 'value'}, 'ttl': 888}
]}
result = self.client.post('/queues/%s/messages' % queue_name,
data=sample_messages)
self.assertEqual(201, result.status_code)
claim_metadata = {'ttl': 100, 'grace': 300}
result = self.client.post('/queues/%s/claims' % queue_name,
data=claim_metadata)
self.assertEqual(201, result.status_code)
response = self.client.get('/health')
self.assertEqual(200, response.status_code)
health = response.json()
self.assertTrue(health['catalog_reachable'])
self.assertTrue(health[pool_name]['storage_reachable'])
op_status = health[pool_name]['operation_status']
for op in op_status.keys():
self.assertTrue(op_status[op]['succeeded'])
message_volume = health[pool_name]['message_volume']
self.assertEqual(2, message_volume['claimed'])
self.assertEqual(0, message_volume['free'])
self.assertEqual(2, message_volume['total'])
zaqar/tests/functional/wsgi/v1_1/test_messages.py
-527
View File
@@ -1,527 +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.
import uuid
import ddt
from oslo_serialization import jsonutils
from zaqar.common import consts
from zaqar.tests.functional import base
from zaqar.tests.functional import helpers
@ddt.ddt
class TestMessages(base.V1_1FunctionalTestBase):
"""Message Tests Specific to V1.1."""
server_class = base.ZaqarServer
def setUp(self):
super(TestMessages, self).setUp()
self.queue = uuid.uuid1() # Generate a random queue ID
self.queue_url = ("{url}/{version}/queues/{queue}".format(
url=self.cfg.zaqar.url,
version="v1.1",
queue=self.queue))
self.headers = helpers.create_zaqar_headers(self.cfg)
self.client.headers = self.headers
self.client.put(self.queue_url) # Create the queue
self.message_url = self.queue_url + '/messages'
self.client.set_base_url(self.message_url)
def tearDown(self):
self.client.delete(self.queue_url) # Remove the queue
super(TestMessages, self).tearDown()
def _post_large_bulk_insert(self, offset):
"""Insert just under than max allowed messages."""
message1 = {"body": '', "ttl": 300}
message2 = {"body": '', "ttl": 120}
doc = {'messages': [message1, message2]}
overhead = len(jsonutils.dumps(doc))
half_size = (self.limits.max_messages_post_size - overhead) // 2
message1['body'] = helpers.generate_random_string(half_size)
message2['body'] = helpers.generate_random_string(half_size + offset)
return self.client.post(data=doc)
def test_message_single_insert(self):
"""Insert Single Message into the Queue.
This test also verifies that claimed messages are
retuned (or not) depending on the include_claimed flag.
"""
doc = helpers.create_message_body_v1_1(messagecount=1)
result = self.client.post(data=doc)
self.assertEqual(201, result.status_code)
response_headers = set(result.headers.keys())
self.assertIsSubset(self.headers_response_with_body, response_headers)
# GET on posted message
href = result.json()['resources'][0]
url = self.cfg.zaqar.url + href
result = self.client.get(url)
self.assertEqual(200, result.status_code)
# Compare message metadata
result_body = result.json()['body']
posted_metadata = doc['messages'][0]['body']
self.assertEqual(posted_metadata, result_body)
# Post a claim & verify the include_claimed flag.
url = self.queue_url + '/claims'
doc = {"ttl": 300, "grace": 100}
result = self.client.post(url, data=doc)
self.assertEqual(201, result.status_code)
params = {'include_claimed': True,
'echo': True}
result = self.client.get(params=params)
self.assertEqual(200, result.status_code)
response_message_body = result.json()["messages"][0]["body"]
self.assertEqual(posted_metadata, response_message_body)
# By default, include_claimed = false
result = self.client.get(self.message_url)
self.assertEqual(200, result.status_code)
test_message_single_insert.tags = ['smoke', 'positive']
def test_message_bulk_insert(self):
"""Bulk Insert Messages into the Queue."""
message_count = self.limits.max_messages_per_page
doc = helpers.create_message_body_v1_1(messagecount=message_count)
result = self.client.post(data=doc)
self.assertEqual(201, result.status_code)
# GET on posted messages
location = result.headers['location']
url = self.cfg.zaqar.url + location
result = self.client.get(url)
self.assertEqual(200, result.status_code)
# Verify that the response json schema matches the expected schema
self.assertSchema(result.json(), consts.MESSAGE_GET_MANY)
self.skipTest('Bug #1273335 - Get set of messages returns wrong hrefs '
'(happens randomly)')
# Compare message metadata
result_body = [msg['body'] for msg in result.json()['messages']]
result_body.sort()
posted_metadata = [msg['body'] for msg in doc['messages']]
posted_metadata.sort()
self.assertEqual(posted_metadata, result_body)
test_message_bulk_insert.tags = ['smoke', 'positive']
def test_message_default_ttl(self):
"""Insert Single Message into the Queue using the default TTL."""
doc = helpers.create_message_body_v1_1(messagecount=1,
default_ttl=True)
result = self.client.post(data=doc)
self.assertEqual(201, result.status_code)
# GET on posted message
href = result.json()['resources'][0]
url = self.cfg.zaqar.url + href
result = self.client.get(url)
self.assertEqual(200, result.status_code)
# Compare message metadata
default_ttl = result.json()['ttl']
self.assertEqual(self.resource_defaults.message_ttl, default_ttl)
test_message_default_ttl.tags = ['smoke', 'positive']
@ddt.data({}, {'limit': 5})
def test_get_message(self, params):
"""Get Messages."""
# Note(abettadapur): This will now return 200s and [].
# Needs to be addressed when feature patch goes in
self.skipTest("Not supported")
expected_msg_count = params.get('limit',
self.limits.max_messages_per_page)
# Test Setup
doc = helpers.create_message_body_v1_1(
messagecount=self.limits.max_messages_per_page)
result = self.client.post(data=doc)
self.assertEqual(201, result.status_code)
url = ''
params['echo'] = True
# Follow the hrefs & perform GET, till the end of messages i.e. http
# 204
while result.status_code in [201, 200]:
result = self.client.get(url, params=params)
self.assertIn(result.status_code, [200, 204])
if result.status_code == 200:
actual_msg_count = len(result.json()['messages'])
self.assertMessageCount(actual_msg_count, expected_msg_count)
href = result.json()['links'][0]['href']
url = self.cfg.zaqar.url + href
self.assertEqual(204, result.status_code)
test_get_message.tags = ['smoke', 'positive']
def test_message_delete(self):
"""Delete Message."""
# Test Setup
doc = helpers.create_message_body_v1_1(messagecount=1)
result = self.client.post(data=doc)
self.assertEqual(201, result.status_code)
# Delete posted message
href = result.json()['resources'][0]
url = self.cfg.zaqar.url + href
result = self.client.delete(url)
self.assertEqual(204, result.status_code)
result = self.client.get(url)
self.assertEqual(404, result.status_code)
test_message_delete.tags = ['smoke', 'positive']
def test_message_bulk_delete(self):
"""Bulk Delete Messages."""
doc = helpers.create_message_body_v1_1(messagecount=10)
result = self.client.post(data=doc)
self.assertEqual(201, result.status_code)
# Delete posted messages
location = result.headers['Location']
url = self.cfg.zaqar.url + location
result = self.client.delete(url)
self.assertEqual(204, result.status_code)
result = self.client.get(url)
self.assertEqual(404, result.status_code)
test_message_bulk_delete.tags = ['smoke', 'positive']
def test_message_delete_nonexisting(self):
"""Delete non-existing Messages."""
result = self.client.delete('/non-existing')
self.assertEqual(204, result.status_code)
test_message_delete_nonexisting.tags = ['negative']
def test_message_partial_delete(self):
"""Delete Messages will be partially successful."""
doc = helpers.create_message_body_v1_1(messagecount=3)
result = self.client.post(data=doc)
self.assertEqual(201, result.status_code)
# Delete posted message
location = result.headers['Location']
url = self.cfg.zaqar.url + location
url += ',nonexisting'
result = self.client.delete(url)
self.assertEqual(204, result.status_code)
test_message_partial_delete.tags = ['negative']
@ddt.data(5, 1)
def test_messages_pop(self, limit=5):
"""Pop messages from a queue."""
doc = helpers.create_message_body_v1_1(messagecount=limit)
result = self.client.post(data=doc)
self.assertEqual(201, result.status_code)
# Pop messages
url = self.message_url + '?pop=' + str(limit)
result = self.client.delete(url)
self.assertEqual(200, result.status_code)
params = {'echo': True}
result = self.client.get(self.message_url, params=params)
self.assertEqual(200, result.status_code)
messages = result.json()['messages']
self.assertEqual([], messages)
test_messages_pop.tags = ['smoke', 'positive']
@ddt.data(10000000, 0, -1)
def test_messages_pop_invalid(self, limit):
"""Pop messages from a queue."""
doc = helpers.create_message_body_v1_1(
messagecount=self.limits.max_messages_per_page)
result = self.client.post(data=doc)
self.assertEqual(201, result.status_code)
# Pop messages
url = self.message_url + '?pop=' + str(limit)
result = self.client.delete(url)
self.assertEqual(400, result.status_code)
params = {'echo': True}
result = self.client.get(self.message_url, params=params)
self.assertEqual(200, result.status_code)
messages = result.json()['messages']
self.assertNotEqual(messages, [])
test_messages_pop_invalid.tags = ['smoke', 'negative']
def test_messages_delete_pop_and_id(self):
"""Delete messages with pop & id params in the request."""
doc = helpers.create_message_body_v1_1(
messagecount=1)
result = self.client.post(data=doc)
self.assertEqual(201, result.status_code)
location = result.headers['Location']
# Pop messages
url = self.cfg.zaqar.url + location + '&pop=1'
result = self.client.delete(url)
self.assertEqual(400, result.status_code)
params = {'echo': True}
result = self.client.get(self.message_url, params=params)
self.assertEqual(200, result.status_code)
messages = result.json()['messages']
self.assertNotEqual(messages, [])
test_messages_delete_pop_and_id.tags = ['smoke', 'negative']
def test_messages_pop_empty_queue(self):
"""Pop messages from an empty queue."""
url = self.message_url + '?pop=2'
result = self.client.delete(url)
self.assertEqual(200, result.status_code)
messages = result.json()['messages']
self.assertEqual([], messages)
test_messages_pop_empty_queue.tags = ['smoke', 'positive']
def test_messages_pop_one(self):
"""Pop single messages from a queue."""
doc = helpers.create_message_body_v1_1(
messagecount=self.limits.max_messages_per_page)
result = self.client.post(data=doc)
self.assertEqual(201, result.status_code)
# Pop Single Message
url = self.message_url + '?pop=1'
result = self.client.delete(url)
self.assertEqual(200, result.status_code)
# Get messages from the queue & verify message count
params = {'echo': True, 'limit': self.limits.max_messages_per_page}
result = self.client.get(self.message_url, params=params)
self.assertEqual(200, result.status_code)
expected_msg_count = self.limits.max_messages_per_page - 1
actual_msg_count = len(result.json()['messages'])
self.assertEqual(expected_msg_count, actual_msg_count)
test_messages_pop_one.tags = ['smoke', 'positive']
def test_message_partial_get(self):
"""Get Messages will be partially successful."""
doc = helpers.create_message_body_v1_1(messagecount=3)
result = self.client.post(data=doc)
self.assertEqual(201, result.status_code)
# Get posted message and a nonexisting message
location = result.headers['Location']
url = self.cfg.zaqar.url + location
url += ',nonexisting'
result = self.client.get(url)
self.assertEqual(200, result.status_code)
test_message_partial_get.tags = ['negative']
@ddt.data(-10, -1, 0)
def test_message_bulk_insert_large_bodies(self, offset):
"""Insert just under than max allowed messages."""
result = self._post_large_bulk_insert(offset)
self.assertEqual(201, result.status_code)
test_message_bulk_insert_large_bodies.tags = ['positive']
@ddt.data(1, 10)
def test_message_bulk_insert_large_bodies_(self, offset):
"""Insert just under than max allowed messages."""
result = self._post_large_bulk_insert(offset)
self.assertEqual(400, result.status_code)
test_message_bulk_insert_large_bodies_.tags = ['negative']
def test_message_bulk_insert_oversized(self):
"""Insert more than max allowed size."""
doc = '[{{"body": "{0}", "ttl": 300}}, {{"body": "{1}", "ttl": 120}}]'
overhead = len(doc.format('', ''))
half_size = (self.limits.max_messages_post_size - overhead) // 2
doc = doc.format(helpers.generate_random_string(half_size),
helpers.generate_random_string(half_size + 1))
result = self.client.post(data=doc)
self.assertEqual(400, result.status_code)
test_message_bulk_insert_oversized.tags = ['negative']
@ddt.data(10000000000000000000, -100, 0, 30, -10000000000000000000)
def test_message_get_invalid_limit(self, limit):
"""Get Messages with invalid value for limit.
Allowed values for limit are 0 < limit <= 20(configurable).
"""
params = {'limit': limit}
result = self.client.get(params=params)
self.assertEqual(400, result.status_code)
test_message_get_invalid_limit.tags = ['negative']
def test_message_bulk_delete_negative(self):
"""Delete more messages than allowed in a single request.
By default, max messages that can be deleted in a single
request is 20.
"""
url = (self.message_url + '?ids='
+ ','.join(str(i) for i in
range(self.limits.max_messages_per_page + 1)))
result = self.client.delete(url)
self.assertEqual(400, result.status_code)
test_message_bulk_delete_negative.tags = ['negative']
def test_message_bulk_get_negative(self):
"""GET more messages by id than allowed in a single request.
By default, max messages that can be fetched in a single
request is 20.
"""
url = (self.message_url + '?ids='
+ ','.join(str(i) for i in
range(self.limits.max_messages_per_page + 1)))
result = self.client.get(url)
self.assertEqual(400, result.status_code)
test_message_bulk_get_negative.tags = ['negative']
def test_get_messages_malformed_marker(self):
"""Get messages with non-existing marker."""
url = self.message_url + '?marker=invalid'
result = self.client.get(url, headers=self.headers)
self.assertEqual(200, result.status_code)
self.assertSchema(result.json(), 'message_list')
test_get_messages_malformed_marker.tags = ['negative']
@ddt.data(None, '1234', 'aa2-bb3',
'103e09c6-31b7-11e3-86bc-b8ca3ad0f5d81',
'103e09c6-31b7-11e3-86bc-b8ca3ad0f5d')
def test_get_messages_invalid_client_id(self, client_id):
"""Get messages with invalid client id."""
url = self.message_url
header = helpers.create_zaqar_headers(self.cfg)
header['Client-ID'] = client_id
result = self.client.get(url, headers=header)
self.assertEqual(400, result.status_code)
test_get_messages_invalid_client_id.tags = ['negative']
def test_query_non_existing_message(self):
"""Get Non Existing Message."""
path = '/non-existing-message'
result = self.client.get(path)
self.assertEqual(404, result.status_code)
test_query_non_existing_message.tags = ['negative']
def test_query_non_existing_message_set(self):
"""Get Set of Non Existing Messages."""
path = '?ids=not_there1,not_there2'
result = self.client.get(path)
self.assertEqual(404, result.status_code)
test_query_non_existing_message_set.tags = ['negative']
def test_delete_non_existing_message(self):
"""Delete Non Existing Message."""
path = '/non-existing-message'
result = self.client.delete(path)
self.assertEqual(204, result.status_code)
test_delete_non_existing_message.tags = ['negative']
def test_message_bad_header_single_insert(self):
"""Insert Single Message into the Queue.
This should fail because of the lack of a Client-ID header
"""
self.skipTest("Not supported")
del self.client.headers["Client-ID"]
doc = helpers.create_message_body_v1_1(messagecount=1)
result = self.client.post(data=doc)
self.assertEqual(400, result.status_code)
zaqar/tests/functional/wsgi/v1_1/test_pools.py
-234
View File
@@ -1,234 +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.
import ddt
from zaqar import tests as testing
from zaqar.tests.functional import base
from zaqar.tests.functional import helpers
@ddt.ddt
class TestPools(base.V1_1FunctionalTestBase):
server_class = base.ZaqarAdminServer
config_file = 'wsgi_mongodb_pooled.conf'
@testing.requires_mongodb
def setUp(self):
super(TestPools, self).setUp()
self.pool_url = ("{url}/{version}/pools".format(
url=self.cfg.zaqar.url,
version="v1.1"
))
self.cfg.zaqar.version = "v1.1"
self.headers = helpers.create_zaqar_headers(self.cfg)
self.client.headers = self.headers
self.client.set_base_url(self.pool_url)
@ddt.data(
{
'name': "newpool",
'weight': 10
}
)
def test_insert_pool(self, params):
"""Test the registering of one pool."""
doc = helpers.create_pool_body(
weight=params.get('weight', 10),
uri=self.mongodb_url
)
pool_name = params.get('name', "newpool")
self.addCleanup(self.client.delete, url='/' + pool_name)
result = self.client.put('/' + pool_name, data=doc)
self.assertEqual(201, result.status_code)
# Test existence
result = self.client.get('/' + pool_name)
self.assertEqual(200, result.status_code)
@ddt.data(
{
'name': "newpool",
'weight': 10
}
)
def test_pool_details(self, params):
"""Get the details of a pool. Assert the respective schema."""
doc = helpers.create_pool_body(
weight=params.get('weight', 10),
uri=self.mongodb_url
)
pool_name = params.get('name', "newpool")
self.addCleanup(self.client.delete, url='/' + pool_name)
result = self.client.put('/' + pool_name, data=doc)
self.assertEqual(201, result.status_code)
# Test existence
result = self.client.get('/' + pool_name + '?detailed=true')
self.assertEqual(200, result.status_code)
self.assertSchema(result.json(), 'pool_get_detail')
@ddt.data(
{
'name': "newpool",
'weight': 10,
}
)
def test_delete_pool(self, params):
"""Create a pool, then delete it.
Make sure operation is successful.
"""
# Create the pool
doc = helpers.create_pool_body(
weight=params.get('weight', 10),
uri=self.mongodb_url
)
pool_name = params.get('name', "newpool")
result = self.client.put('/' + pool_name, data=doc)
self.assertEqual(201, result.status_code)
# Make sure it exists
result = self.client.get('/' + pool_name)
self.assertEqual(200, result.status_code)
# Delete it
result = self.client.delete('/' + pool_name)
self.assertEqual(204, result.status_code)
@ddt.data(
{
'name': "newpool",
'weight': 10,
}
)
def test_list_pools(self, params):
"""Add a pool. Get the list of all the pools.
Assert respective schema
"""
doc = helpers.create_pool_body(
weight=params.get('weight', 10),
uri=self.mongodb_url
)
pool_name = params.get('name', "newpool")
self.addCleanup(self.client.delete, url='/' + pool_name)
result = self.client.put('/' + pool_name, data=doc)
self.assertEqual(201, result.status_code)
result = self.client.get()
self.assertEqual(200, result.status_code)
self.assertSchema(result.json(), 'pool_list')
@ddt.data(
{
'name': "newpool",
'weight': 10,
}
)
def test_patch_pool(self, params):
"""Create a pool. Issue a patch command,
make sure command was successful. Check details to be sure.
"""
doc = helpers.create_pool_body(
weight=params.get('weight', 10),
uri=self.mongodb_url
)
pool_name = params.get('name', "newpool")
self.addCleanup(self.client.delete, url='/' + pool_name)
result = self.client.put('/' + pool_name, data=doc)
self.assertEqual(201, result.status_code)
# Update that pool
patchdoc = helpers.create_pool_body(
weight=5,
uri=self.mongodb_url
)
result = self.client.patch('/' + pool_name, data=patchdoc)
self.assertEqual(200, result.status_code)
# Get the pool, check update#
result = self.client.get('/' + pool_name)
self.assertEqual(200, result.status_code)
self.assertEqual(5, result.json()["weight"])
@ddt.data(
{
'name': "newpool",
'weight': 10,
}
)
def test_patch_pool_bad_data(self, params):
"""Issue a patch command without a body. Assert 400."""
# create a pool
doc = helpers.create_pool_body(
weight=params.get('weight', 10),
uri=self.mongodb_url
)
pool_name = params.get('name', "newpool")
self.addCleanup(self.client.delete, url='/' + pool_name)
result = self.client.put('/' + pool_name, data=doc)
self.assertEqual(201, result.status_code)
# Update pool with bad post data. Ensure 400
result = self.client.patch('/' + pool_name)
self.assertEqual(400, result.status_code)
@ddt.data(
{
'name': "newpool",
'weight': 10,
}
)
def test_patch_pool_non_exist(self, params):
"""Issue patch command to pool that doesn't exist. Assert 404."""
doc = helpers.create_pool_body(
weight=5,
uri=self.mongodb_url
)
result = self.client.patch('/nonexistpool', data=doc)
self.assertEqual(404, result.status_code)
@ddt.data(
{'name': '\u6c49\u5b57\u6f22\u5b57'},
{'name': 'i' * 65},
{'weight': -1}
)
def test_insert_pool_bad_data(self, params):
"""Create pools with invalid names and weights. Assert 400."""
self.skip("FIXME: https://bugs.launchpad.net/zaqar/+bug/1373486")
doc = helpers.create_pool_body(
weight=params.get('weight', 10),
uri=self.mongodb_url
)
pool_name = params.get('name', "newpool")
self.addCleanup(self.client.delete, url='/' + pool_name)
result = self.client.put('/' + pool_name, data=doc)
self.assertEqual(400, result.status_code)
def test_delete_pool_non_exist(self):
"""Delete a pool that doesn't exist. Assert 404."""
result = self.client.delete('/nonexistpool')
self.assertEqual(204, result.status_code)
zaqar/tests/functional/wsgi/v1_1/test_queues.py
-351
View File
@@ -1,351 +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.
import uuid
import ddt
from zaqar.tests.functional import base
from zaqar.tests.functional import helpers
class NamedBinaryStr(bytes):
"""Wrapper for bytes to facilitate overriding __name__."""
class NamedUnicodeStr(str):
"""Unicode string look-alike to facilitate overriding __name__."""
def __init__(self, value):
self.value = value
def __str__(self):
return self.value
def encode(self, enc):
return self.value.encode(enc)
def __format__(self, formatstr):
"""Workaround for ddt bug.
DDT will always call __format__ even when __name__ exists,
which blows up for Unicode strings under Py2.
"""
return ''
class NamedDict(dict):
"""Wrapper for dict to facilitate overriding __name__."""
def annotated(test_name, test_input):
if isinstance(test_input, dict):
annotated_input = NamedDict(test_input)
elif isinstance(test_input, str):
annotated_input = NamedUnicodeStr(test_input)
else:
annotated_input = NamedBinaryStr(test_input)
setattr(annotated_input, '__name__', test_name)
return annotated_input
@ddt.ddt
class TestInsertQueue(base.V1_1FunctionalTestBase):
"""Tests for Insert queue."""
server_class = base.ZaqarServer
def setUp(self):
super(TestInsertQueue, self).setUp()
self.base_url = '{0}/{1}'.format(self.cfg.zaqar.url,
"v1.1")
self.header = helpers.create_zaqar_headers(self.cfg)
self.headers_response_empty = {'location'}
self.client.set_base_url(self.base_url)
self.client.headers = self.header
@ddt.data('qtestqueue', 'TESTqueue', 'hyphen-name', '_undersore',
annotated('test_insert_queue_long_name', 'i' * 64))
def test_insert_queue(self, queue_name):
"""Create Queue."""
self.url = self.base_url + '/queues/' + queue_name
self.addCleanup(self.client.delete, self.url)
result = self.client.put(self.url)
self.assertEqual(201, result.status_code)
response_headers = set(result.headers.keys())
self.assertIsSubset(self.headers_response_empty, response_headers)
test_insert_queue.tags = ['positive', 'smoke']
@ddt.data(annotated('test_insert_queue_non_ascii_name',
'\u6c49\u5b57\u6f22\u5b57'),
'@$@^qw',
annotated('test_insert_queue_invalid_name_length', 'i' * 65))
def test_insert_queue_invalid_name(self, queue_name):
"""Create Queue."""
self.url = self.base_url + '/queues/' + queue_name
self.addCleanup(self.client.delete, self.url)
result = self.client.put(self.url)
self.assertEqual(400, result.status_code)
test_insert_queue_invalid_name.tags = ['negative']
def test_insert_queue_header_plaintext(self):
"""Insert Queue with 'Accept': 'plain/text'."""
path = '/queues/plaintextheader'
self.addCleanup(self.client.delete, path)
header = {"Accept": 'plain/text'}
result = self.client.put(path, headers=header)
self.assertEqual(406, result.status_code)
test_insert_queue_header_plaintext.tags = ['negative']
def test_insert_queue_header_asterisk(self):
"""Insert Queue with 'Accept': '*/*'."""
path = '/queues/asteriskinheader'
headers = {'Accept': '*/*',
'Client-ID': str(uuid.uuid4()),
'X-Project-ID': '518b51ea133c4facadae42c328d6b77b'}
self.addCleanup(self.client.delete, url=path, headers=headers)
result = self.client.put(path, headers=headers)
self.assertEqual(201, result.status_code)
test_insert_queue_header_asterisk.tags = ['positive']
def test_insert_queue_with_metadata(self):
"""Insert queue with a non-empty request body."""
self.url = self.base_url + '/queues/hasmetadata'
doc = {"queue": "Has Metadata"}
self.addCleanup(self.client.delete, self.url)
result = self.client.put(self.url, data=doc)
self.assertEqual(201, result.status_code)
self.url = self.base_url + '/queues/hasmetadata'
result = self.client.get(self.url)
self.assertEqual(200, result.status_code)
self.assertEqual({"queue": "Has Metadata"}, result.json())
test_insert_queue_with_metadata.tags = ['negative']
def tearDown(self):
super(TestInsertQueue, self).tearDown()
@ddt.ddt
class TestQueueMisc(base.V1_1FunctionalTestBase):
server_class = base.ZaqarServer
def setUp(self):
super(TestQueueMisc, self).setUp()
self.base_url = self.cfg.zaqar.url
self.client.set_base_url(self.base_url)
self.queue_url = self.base_url + ('/{0}/queues/{1}'
.format("v1.1", uuid.uuid1()))
def test_list_queues(self):
"""List Queues."""
self.client.put(self.queue_url)
self.addCleanup(self.client.delete, self.queue_url)
result = self.client.get('/{0}/queues'
.format("v1.1"))
self.assertEqual(200, result.status_code)
self.assertSchema(result.json(), 'queue_list')
test_list_queues.tags = ['smoke', 'positive']
def test_list_queues_detailed(self):
"""List Queues with detailed = True."""
self.client.put(self.queue_url)
self.addCleanup(self.client.delete, self.queue_url)
params = {'detailed': True}
result = self.client.get('/{0}/queues'
.format("v1.1"),
params=params)
self.assertEqual(200, result.status_code)
self.assertSchema(result.json(), 'queue_list')
response_keys = result.json()['queues'][0].keys()
self.assertIn('metadata', response_keys)
test_list_queues_detailed.tags = ['smoke', 'positive']
@ddt.data(0, -1, 1001)
def test_list_queue_invalid_limit(self, limit):
"""List Queues with a limit value that is not allowed."""
params = {'limit': limit}
result = self.client.get('/{0}/queues'
.format("v1.1"),
params=params)
self.assertEqual(400, result.status_code)
test_list_queue_invalid_limit.tags = ['negative']
def test_check_queue_exists(self):
"""Checks if queue exists."""
self.client.put(self.queue_url)
self.addCleanup(self.client.delete, self.queue_url)
result = self.client.head(self.queue_url)
self.assertEqual(405, result.status_code)
test_check_queue_exists.tags = ['negative']
def test_get_queue_malformed_marker(self):
"""List queues with invalid marker."""
path = '/{0}/queues?marker=zzz'.format("v1.1")
result = self.client.get(path)
self.assertEqual(200, result.status_code)
test_get_queue_malformed_marker.tags = ['negative']
def test_get_stats_empty_queue(self):
"""Get queue stats on an empty queue."""
result = self.client.put(self.queue_url)
self.addCleanup(self.client.delete, self.queue_url)
self.assertEqual(201, result.status_code)
stats_url = self.queue_url + '/stats'
# Get stats on an empty queue
result = self.client.get(stats_url)
self.assertEqual(200, result.status_code)
expected_response = {'messages':
{'claimed': 0, 'total': 0, 'free': 0}}
self.assertEqual(expected_response, result.json())
test_get_stats_empty_queue.tags = ['positive']
@ddt.data(0, 1)
def test_get_queue_stats_claimed(self, claimed):
"""Get stats on a queue."""
result = self.client.put(self.queue_url)
self.addCleanup(self.client.delete, self.queue_url)
self.assertEqual(201, result.status_code)
# Post Messages to the test queue
doc = helpers.create_message_body_v1_1(
messagecount=self.limits.max_messages_per_claim_or_pop)
message_url = self.queue_url + '/messages'
result = self.client.post(message_url, data=doc)
self.assertEqual(201, result.status_code)
if claimed > 0:
claim_url = self.queue_url + '/claims?limit=' + str(claimed)
doc = {'ttl': 300, 'grace': 300}
result = self.client.post(claim_url, data=doc)
self.assertEqual(201, result.status_code)
# Get stats on the queue.
stats_url = self.queue_url + '/stats'
result = self.client.get(stats_url)
self.assertEqual(200, result.status_code)
self.assertQueueStats(result.json(), claimed)
test_get_queue_stats_claimed.tags = ['positive']
def test_ping_queue(self):
pass
def tearDown(self):
super(TestQueueMisc, self).tearDown()
class TestQueueNonExisting(base.V1_1FunctionalTestBase):
"""Test Actions on non existing queue."""
server_class = base.ZaqarServer
def setUp(self):
super(TestQueueNonExisting, self).setUp()
if self.cfg.version != "v1":
self.skipTest("Not Supported")
self.base_url = '{0}/{1}'.format(self.cfg.zaqar.url,
"v1.1")
self.queue_url = (self.base_url +
'/queues/0a5b1b85-4263-11e3-b034-28cfe91478b9')
self.client.set_base_url(self.queue_url)
self.header = helpers.create_zaqar_headers(self.cfg)
self.headers_response_empty = {'location'}
self.header = helpers.create_zaqar_headers(self.cfg)
def test_get_stats(self):
"""Get stats on non existing Queue."""
result = self.client.get('/stats')
self.assertEqual(200, result.status_code)
self.assertEqual([], result.json())
def test_get_metadata(self):
"""Get metadata on non existing Queue."""
result = self.client.get('/')
self.assertEqual(200, result.status_code)
self.assertEqual([], result.json())
def test_get_messages(self):
"""Get messages on non existing Queue."""
result = self.client.get('/messages')
self.assertEqual(200, result.status_code)
self.assertEqual([], result.json())
def test_post_messages(self):
"""Post messages to a non existing Queue."""
doc = [{"ttl": 200, "body": {"Home": ""}}]
result = self.client.post('/messages', data=doc)
self.assertEqual(201, result.status_code)
# check existence of queue
result = self.client.get()
self.assertEqual(200, result.status_code)
self.assertNotEqual([], result.json())
def test_claim_messages(self):
"""Claim messages from a non existing Queue."""
doc = {"ttl": 200, "grace": 300}
result = self.client.post('/claims', data=doc)
self.assertEqual(200, result.status_code)
self.assertEqual([], result.json())
def test_delete_queue(self):
"""Delete non existing Queue."""
result = self.client.delete()
self.assertEqual(204, result.status_code)
zaqar/tests/unit/transport/wsgi/test_version.py
+1 -18
View File
@@ -19,23 +19,6 @@ from oslo_serialization import jsonutils
from zaqar.tests.unit.transport.wsgi import base
EXPECTED_VERSIONS = [
{
'id': '1.1',
'status': 'DEPRECATED',
'updated': '2016-7-29T02:22:47Z',
'media-types': [
{
'base': 'application/json',
'type': 'application/vnd.openstack.messaging-v1_1+json'
}
],
'links': [
{
'href': '/v1.1/',
'rel': 'self'
}
]
},
{
'id': '2',
'status': 'CURRENT',
@@ -65,5 +48,5 @@ class TestVersion(base.TestBase):
versions = jsonutils.loads(response[0])['versions']
self.assertEqual(falcon.HTTP_300, self.srmock.status)
self.assertEqual(2, len(versions))
self.assertEqual(1, len(versions))
self.assertEqual(EXPECTED_VERSIONS, versions)
zaqar/transport/wsgi/driver.py
-3
View File
@@ -32,7 +32,6 @@ from zaqar.transport.middleware import auth
from zaqar.transport.middleware import cors
from zaqar.transport.middleware import profile
from zaqar.transport import validation
from zaqar.transport.wsgi import v1_1
from zaqar.transport.wsgi import v2_0
from zaqar.transport.wsgi import version
@@ -108,14 +107,12 @@ class Driver(transport.DriverBase):
"""Initialize hooks and URI routes to resources."""
catalog = [
('/v1.1', v1_1.public_endpoints(self, self._conf)),
('/v2', v2_0.public_endpoints(self, self._conf)),
('/', [('', version.Resource())])
]
if self._conf.admin_mode:
catalog.extend([
('/v1.1', v1_1.private_endpoints(self, self._conf)),
('/v2', v2_0.private_endpoints(self, self._conf)),
])
zaqar/transport/wsgi/v1_1/__init__.py
-129
View File
@@ -1,129 +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.
from oslo_log import log as logging
from zaqar.common import decorators
from zaqar.transport.wsgi.v1_1 import claims
from zaqar.transport.wsgi.v1_1 import flavors
from zaqar.transport.wsgi.v1_1 import health
from zaqar.transport.wsgi.v1_1 import homedoc
from zaqar.transport.wsgi.v1_1 import messages
from zaqar.transport.wsgi.v1_1 import ping
from zaqar.transport.wsgi.v1_1 import pools
from zaqar.transport.wsgi.v1_1 import queues
from zaqar.transport.wsgi.v1_1 import stats
LOG = logging.getLogger(__name__)
VERSION = {
'id': '1.1',
'status': 'DEPRECATED',
'updated': '2016-7-29T02:22:47Z',
'media-types': [
{
'base': 'application/json',
'type': 'application/vnd.openstack.messaging-v1_1+json'
}
],
'links': [
{
'href': '/v1.1/',
'rel': 'self'
}
]
}
@decorators.api_version_manager(VERSION)
def public_endpoints(driver, conf):
queue_controller = driver._storage.queue_controller
message_controller = driver._storage.message_controller
claim_controller = driver._storage.claim_controller
defaults = driver._defaults
return [
# Home
('/',
homedoc.Resource(conf)),
# Queues Endpoints
('/queues',
queues.CollectionResource(driver._validate,
queue_controller)),
('/queues/{queue_name}',
queues.ItemResource(driver._validate,
queue_controller,
message_controller)),
('/queues/{queue_name}/stats',
stats.Resource(queue_controller)),
# Messages Endpoints
('/queues/{queue_name}/messages',
messages.CollectionResource(driver._wsgi_conf,
driver._validate,
message_controller,
queue_controller,
defaults.message_ttl)),
('/queues/{queue_name}/messages/{message_id}',
messages.ItemResource(message_controller)),
# Claims Endpoints
('/queues/{queue_name}/claims',
claims.CollectionResource(driver._wsgi_conf,
driver._validate,
claim_controller,
defaults.claim_ttl,
defaults.claim_grace)),
('/queues/{queue_name}/claims/{claim_id}',
claims.ItemResource(driver._wsgi_conf,
driver._validate,
claim_controller,
defaults.claim_ttl,
defaults.claim_grace)),
# Ping
('/ping',
ping.Resource(driver._storage))
]
@decorators.api_version_manager(VERSION)
def private_endpoints(driver, conf):
catalogue = [
# Health
('/health',
health.Resource(driver._storage)),
]
if conf.pooling:
pools_controller = driver._control.pools_controller
flavors_controller = driver._control.flavors_controller
catalogue.extend([
('/pools',
pools.Listing(pools_controller)),
('/pools/{pool}',
pools.Resource(pools_controller)),
('/flavors',
flavors.Listing(flavors_controller)),
('/flavors/{flavor}',
flavors.Resource(flavors_controller)),
])
return catalogue
zaqar/transport/wsgi/v1_1/claims.py
-199
View File
@@ -1,199 +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 falcon
from oslo_log import log as logging
from zaqar.common import decorators
from zaqar.i18n import _
from zaqar.storage import errors as storage_errors
from zaqar.transport import utils
from zaqar.transport import validation
from zaqar.transport.wsgi import errors as wsgi_errors
from zaqar.transport.wsgi import utils as wsgi_utils
LOG = logging.getLogger(__name__)
class CollectionResource(object):
__slots__ = (
'_claim_controller',
'_validate',
'_claim_post_spec',
'_default_meta',
)
def __init__(self, wsgi_conf, validate, claim_controller,
default_claim_ttl, default_grace_ttl):
self._claim_controller = claim_controller
self._validate = validate
self._claim_post_spec = (
('ttl', int, default_claim_ttl),
('grace', int, default_grace_ttl),
)
# NOTE(kgriffs): Create this once up front, rather than creating
# a new dict every time, for the sake of performance.
self._default_meta = {
'ttl': default_claim_ttl,
'grace': default_grace_ttl,
}
@decorators.TransportLog("Claims collection")
def on_post(self, req, resp, project_id, queue_name):
# Check for an explicit limit on the # of messages to claim
limit = req.get_param_as_int('limit')
claim_options = {} if limit is None else {'limit': limit}
# NOTE(kgriffs): Clients may or may not actually include the
# Content-Length header when the body is empty; the following
# check works for both 0 and None.
if not req.content_length:
# No values given, so use defaults
metadata = self._default_meta
else:
# Read claim metadata (e.g., TTL) and raise appropriate
# HTTP errors as needed.
document = wsgi_utils.deserialize(req.stream, req.content_length)
metadata = wsgi_utils.sanitize(document, self._claim_post_spec)
# Claim some messages
try:
self._validate.claim_creation(metadata, limit=limit)
cid, msgs = self._claim_controller.create(
queue_name,
metadata=metadata,
project=project_id,
**claim_options)
# Buffer claimed messages
# TODO(kgriffs): optimize, along with serialization (below)
resp_msgs = list(msgs)
except validation.ValidationFailed as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPBadRequestAPI(str(ex))
except Exception:
description = _('Claim could not be created.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
# Serialize claimed messages, if any. This logic assumes
# the storage driver returned well-formed messages.
if len(resp_msgs) != 0:
base_path = req.path.rpartition('/')[0]
resp_msgs = [wsgi_utils.format_message_v1_1(msg, base_path, cid)
for msg in resp_msgs]
resp.location = req.path + '/' + cid
resp.text = utils.to_json({'messages': resp_msgs})
resp.status = falcon.HTTP_201
else:
resp.status = falcon.HTTP_204
class ItemResource(object):
__slots__ = ('_claim_controller', '_validate', '_claim_patch_spec')
def __init__(self, wsgi_conf, validate, claim_controller,
default_claim_ttl, default_grace_ttl):
self._claim_controller = claim_controller
self._validate = validate
self._claim_patch_spec = (
('ttl', int, default_claim_ttl),
('grace', int, default_grace_ttl),
)
@decorators.TransportLog("Claim item")
def on_get(self, req, resp, project_id, queue_name, claim_id):
try:
meta, msgs = self._claim_controller.get(
queue_name,
claim_id=claim_id,
project=project_id)
# Buffer claimed messages
# TODO(kgriffs): Optimize along with serialization (see below)
meta['messages'] = list(msgs)
except storage_errors.DoesNotExist as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPNotFound(str(ex))
except Exception:
description = _('Claim could not be queried.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
# Serialize claimed messages
# TODO(kgriffs): Optimize
base_path = req.path.rsplit('/', 2)[0]
meta['messages'] = [wsgi_utils.format_message_v1_1(msg, base_path,
claim_id)
for msg in meta['messages']]
meta['href'] = req.path
del meta['id']
resp.text = utils.to_json(meta)
# status defaults to 200
@decorators.TransportLog("Claim item")
def on_patch(self, req, resp, project_id, queue_name, claim_id):
# Read claim metadata (e.g., TTL) and raise appropriate
# HTTP errors as needed.
document = wsgi_utils.deserialize(req.stream, req.content_length)
metadata = wsgi_utils.sanitize(document, self._claim_patch_spec)
try:
self._validate.claim_updating(metadata)
self._claim_controller.update(queue_name,
claim_id=claim_id,
metadata=metadata,
project=project_id)
resp.status = falcon.HTTP_204
except validation.ValidationFailed as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPBadRequestAPI(str(ex))
except storage_errors.DoesNotExist as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPNotFound(str(ex))
except Exception:
description = _('Claim could not be updated.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
@decorators.TransportLog("Claim item")
def on_delete(self, req, resp, project_id, queue_name, claim_id):
try:
self._claim_controller.delete(queue_name,
claim_id=claim_id,
project=project_id)
resp.status = falcon.HTTP_204
except Exception:
description = _('Claim could not be deleted.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
zaqar/transport/wsgi/v1_1/flavors.py
-203
View File
@@ -1,203 +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.
import falcon
import jsonschema
from oslo_log import log
from zaqar.common.api.schemas.v1_1 import flavors as schema
from zaqar.common import utils as common_utils
from zaqar.i18n import _
from zaqar.storage import errors
from zaqar.transport import utils as transport_utils
from zaqar.transport.wsgi import errors as wsgi_errors
from zaqar.transport.wsgi import utils as wsgi_utils
LOG = log.getLogger(__name__)
class Listing(object):
"""A resource to list registered flavors
:param flavors_controller: means to interact with storage
"""
def __init__(self, flavors_controller):
self._ctrl = flavors_controller
def on_get(self, request, response, project_id):
"""Returns a flavor listing as objects embedded in an object:
::
{
"flavors": [
{"href": "", "capabilities": {}, "pool": ""},
...
],
"links": [
{"rel": "next", "href": ""},
...
]
}
:returns: HTTP | 200
"""
LOG.debug('LIST flavors for project_id %s', project_id)
store = {}
request.get_param('marker', store=store)
request.get_param_as_int('limit', store=store)
request.get_param_as_bool('detailed', store=store)
cursor = self._ctrl.list(project=project_id, **store)
flavors = list(next(cursor))
results = {'links': []}
if flavors:
store['marker'] = next(cursor)
for entry in flavors:
entry['href'] = request.path + '/' + entry['name']
results['links'] = [
{
'rel': 'next',
'href': request.path + falcon.to_query_str(store)
}
]
results['flavors'] = flavors
response.text = transport_utils.to_json(results)
response.status = falcon.HTTP_200
class Resource(object):
"""A handler for individual flavor.
:param flavors_controller: means to interact with storage
"""
def __init__(self, flavors_controller):
self._ctrl = flavors_controller
validator_type = jsonschema.Draft4Validator
self._validators = {
'create': validator_type(schema.create),
'capabilities': validator_type(schema.patch_capabilities),
}
def on_get(self, request, response, project_id, flavor):
"""Returns a JSON object for a single flavor entry:
::
{"pool_group": "", capabilities: {...}}
:returns: HTTP | [200, 404]
"""
LOG.debug('GET flavor - name: %s', flavor)
data = None
detailed = request.get_param_as_bool('detailed') or False
try:
data = self._ctrl.get(flavor,
project=project_id,
detailed=detailed)
except errors.FlavorDoesNotExist as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPNotFound(str(ex))
data['href'] = request.path
response.text = transport_utils.to_json(data)
def on_put(self, request, response, project_id, flavor):
"""Registers a new flavor. Expects the following input:
::
{"capabilities": {}}
A capabilities object may also be provided.
:returns: HTTP | [201, 400]
"""
LOG.debug('PUT flavor - name: %s', flavor)
data = wsgi_utils.load(request)
wsgi_utils.validate(self._validators['create'], data)
try:
self._ctrl.create(flavor,
project=project_id,
capabilities=data['capabilities'])
response.status = falcon.HTTP_201
response.location = request.path
except errors.PoolGroupDoesNotExist:
description = (_('Flavor %(flavor)s could not be created. ') %
dict(flavor=flavor))
LOG.exception(description)
raise falcon.HTTPBadRequest(
title=_('Unable to create'), description=description)
def on_delete(self, request, response, project_id, flavor):
"""Deregisters a flavor.
:returns: HTTP | [204]
"""
LOG.debug('DELETE flavor - name: %s', flavor)
self._ctrl.delete(flavor, project=project_id)
response.status = falcon.HTTP_204
def on_patch(self, request, response, project_id, flavor):
"""Allows one to update a flavors's pool and/or capabilities.
This method expects the user to submit a JSON object
containing at least one of: 'pool_group', 'capabilities'. If
none are found, the request is flagged as bad. There is also
strict format checking through the use of
jsonschema. Appropriate errors are returned in each case for
badly formatted input.
:returns: HTTP | [200, 400]
"""
LOG.debug('PATCH flavor - name: %s', flavor)
data = wsgi_utils.load(request)
EXPECT = ('capabilities')
if not any([(field in data) for field in EXPECT]):
LOG.debug('PATCH flavor, bad params')
raise wsgi_errors.HTTPBadRequestBody(
'`capabilities` needs '
'to be specified'
)
for field in EXPECT:
wsgi_utils.validate(self._validators[field], data)
fields = common_utils.fields(data, EXPECT,
pred=lambda v: v is not None)
try:
self._ctrl.update(flavor, project=project_id, **fields)
except errors.FlavorDoesNotExist as ex:
LOG.exception('Flavor "%s" does not exist', flavor)
raise wsgi_errors.HTTPNotFound(str(ex))
zaqar/transport/wsgi/v1_1/health.py
-39
View File
@@ -1,39 +0,0 @@
# Copyright (c) 2014 Rackspace, Inc.
# Copyright 2014 Catalyst IT Ltd.
#
# 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_log import log as logging
from zaqar.i18n import _
from zaqar.transport import utils
from zaqar.transport.wsgi import errors as wsgi_errors
LOG = logging.getLogger(__name__)
class Resource(object):
__slots__ = ('_driver',)
def __init__(self, driver):
self._driver = driver
def on_get(self, req, resp, **kwargs):
try:
resp_dict = self._driver.health()
resp.text = utils.to_json(resp_dict)
except Exception:
description = _('Health status could not be read.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
zaqar/transport/wsgi/v1_1/homedoc.py
-292
View File
@@ -1,292 +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.
from oslo_serialization import jsonutils
# NOTE(kgriffs): http://tools.ietf.org/html/draft-nottingham-json-home-03
JSON_HOME = {
'resources': {
# -----------------------------------------------------------------
# Queues
# -----------------------------------------------------------------
'rel/queues': {
'href-template': '/v1.1/queues{?marker,limit,detailed}',
'href-vars': {
'marker': 'param/marker',
'limit': 'param/queue_limit',
'detailed': 'param/detailed',
},
'hints': {
'allow': ['GET'],
'formats': {
'application/json': {},
},
},
},
'rel/queue': {
'href-template': '/v1.1/queues/{queue_name}',
'href-vars': {
'queue_name': 'param/queue_name',
},
'hints': {
'allow': ['PUT', 'DELETE'],
'formats': {
'application/json': {},
},
},
},
'rel/queue_stats': {
'href-template': '/v1.1/queues/{queue_name}/stats',
'href-vars': {
'queue_name': 'param/queue_name',
},
'hints': {
'allow': ['GET'],
'formats': {
'application/json': {},
},
},
},
# -----------------------------------------------------------------
# Messages
# -----------------------------------------------------------------
'rel/messages': {
'href-template': ('/v1.1/queues/{queue_name}/messages'
'{?marker,limit,echo,include_claimed}'),
'href-vars': {
'queue_name': 'param/queue_name',
'marker': 'param/marker',
'limit': 'param/messages_limit',
'echo': 'param/echo',
'include_claimed': 'param/include_claimed',
},
'hints': {
'allow': ['GET'],
'formats': {
'application/json': {},
},
},
},
'rel/post_messages': {
'href-template': '/v1.1/queues/{queue_name}/messages',
'href-vars': {
'queue_name': 'param/queue_name',
},
'hints': {
'allow': ['POST'],
'formats': {
'application/json': {},
},
'accept-post': ['application/json'],
},
},
'rel/messages_delete': {
'href-template': '/v1.1/queues/{queue_name}/messages{?ids,pop}',
'href-vars': {
'queue_name': 'param/queue_name',
'ids': 'param/ids',
'pop': 'param/pop'
},
'hints': {
'allow': [
'DELETE'
],
'formats': {
'application/json': {}
}
}
},
'rel/message_delete': {
'href-template': '/v1.1/queues/{queue_name}/messages/{message_id}{?claim}', # noqa
'href-vars': {
'queue_name': 'param/queue_name',
'message_id': 'param/message_id',
'claim': 'param/claim_id'
},
'hints': {
'allow': [
'DELETE'
],
'formats': {
'application/json': {}
}
}
},
# -----------------------------------------------------------------
# Claims
# -----------------------------------------------------------------
'rel/claim': {
'href-template': '/v1.1/queues/{queue_name}/claims/{claim_id}',
'href-vars': {
'queue_name': 'param/queue_name',
'claim_id': 'param/claim_id',
},
'hints': {
'allow': ['GET'],
'formats': {
'application/json': {},
},
},
},
'rel/post_claim': {
'href-template': '/v1.1/queues/{queue_name}/claims{?limit}',
'href-vars': {
'queue_name': 'param/queue_name',
'limit': 'param/claim_limit',
},
'hints': {
'allow': ['POST'],
'formats': {
'application/json': {},
},
'accept-post': ['application/json']
},
},
'rel/patch_claim': {
'href-template': '/v1.1/queues/{queue_name}/claims/{claim_id}',
'href-vars': {
'queue_name': 'param/queue_name',
'claim_id': 'param/claim_id',
},
'hints': {
'allow': ['PATCH'],
'formats': {
'application/json': {},
},
'accept-post': ['application/json']
},
},
'rel/delete_claim': {
'href-template': '/v1.1/queues/{queue_name}/claims/{claim_id}',
'href-vars': {
'queue_name': 'param/queue_name',
'claim_id': 'param/claim_id',
},
'hints': {
'allow': ['DELETE'],
'formats': {
'application/json': {},
},
},
},
# -----------------------------------------------------------------
# Ping
# -----------------------------------------------------------------
'rel/ping': {
'href-template': '/v1.1/ping',
'hints': {
'allow': ['GET'],
'formats': {
'application/json': {},
}
}
}
}
}
ADMIN_RESOURCES = {
# -----------------------------------------------------------------
# Pools
# -----------------------------------------------------------------
'rel/pools': {
'href-template': '/v1.1/pools{?detailed,limit,marker}',
'href-vars': {
'detailed': 'param/detailed',
'limit': 'param/pool_limit',
'marker': 'param/marker',
},
'hints': {
'allow': ['GET'],
'formats': {
'application/json': {},
},
},
},
'rel/pool': {
'href-template': '/v1.1/pools/{pool_name}',
'href-vars': {
'pool_name': 'param/pool_name',
},
'hints': {
'allow': ['GET', 'PUT', 'PATCH', 'DELETE'],
'formats': {
'application/json': {},
},
},
},
# -----------------------------------------------------------------
# Flavors
# -----------------------------------------------------------------
'rel/flavors': {
'href-template': '/v1.1/flavors{?detailed,limit,marker}',
'href-vars': {
'detailed': 'param/detailed',
'limit': 'param/flavor_limit',
'marker': 'param/marker',
},
'hints': {
'allow': ['GET'],
'formats': {
'application/json': {},
},
},
},
'rel/flavor': {
'href-template': '/v1.1/flavors/{flavor_name}',
'href-vars': {
'flavor_name': 'param/flavor_name',
},
'hints': {
'allow': ['GET', 'PUT', 'PATCH', 'DELETE'],
'formats': {
'application/json': {},
},
},
},
# -----------------------------------------------------------------
# Health
# -----------------------------------------------------------------
'rel/health': {
'href': '/v1.1/health',
'hints': {
'allow': ['GET'],
'formats': {
'application/json': {},
},
},
},
}
class Resource(object):
def __init__(self, conf):
if conf.admin_mode:
JSON_HOME['resources'].update(ADMIN_RESOURCES)
document = jsonutils.dumps(JSON_HOME, ensure_ascii=False, indent=4)
self.document_utf8 = document.encode('utf-8')
def on_get(self, req, resp, project_id):
resp.data = self.document_utf8
resp.content_type = 'application/json-home'
resp.cache_control = ['max-age=86400']
# status defaults to 200
zaqar/transport/wsgi/v1_1/messages.py
-368
View File
@@ -1,368 +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 falcon
from oslo_log import log as logging
from zaqar.common import decorators
from zaqar.common.transport.wsgi import helpers as wsgi_helpers
from zaqar.i18n import _
from zaqar.storage import errors as storage_errors
from zaqar.transport import utils
from zaqar.transport import validation
from zaqar.transport.wsgi import errors as wsgi_errors
from zaqar.transport.wsgi import utils as wsgi_utils
LOG = logging.getLogger(__name__)
class CollectionResource(object):
__slots__ = (
'_message_controller',
'_queue_controller',
'_wsgi_conf',
'_validate',
'_message_post_spec',
)
def __init__(self, wsgi_conf, validate,
message_controller, queue_controller,
default_message_ttl):
self._wsgi_conf = wsgi_conf
self._validate = validate
self._message_controller = message_controller
self._queue_controller = queue_controller
self._message_post_spec = (
('ttl', int, default_message_ttl),
('body', '*', None),
)
# ----------------------------------------------------------------------
# Helpers
# ----------------------------------------------------------------------
def _get_by_id(self, base_path, project_id, queue_name, ids):
"""Returns one or more messages from the queue by ID."""
try:
self._validate.message_listing(limit=len(ids))
messages = self._message_controller.bulk_get(
queue_name,
message_ids=ids,
project=project_id)
except validation.ValidationFailed as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPBadRequestAPI(str(ex))
except Exception:
description = _('Message could not be retrieved.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
# Prepare response
messages = list(messages)
if not messages:
return None
messages = [wsgi_utils.format_message_v1_1(m, base_path, m['claim_id'])
for m in messages]
return {'messages': messages}
def _get(self, req, project_id, queue_name):
client_uuid = wsgi_helpers.get_client_uuid(req)
kwargs = {}
# NOTE(kgriffs): This syntax ensures that
# we don't clobber default values with None.
req.get_param('marker', store=kwargs)
req.get_param_as_int('limit', store=kwargs)
req.get_param_as_bool('echo', store=kwargs)
req.get_param_as_bool('include_claimed', store=kwargs)
try:
self._validate.message_listing(**kwargs)
results = self._message_controller.list(
queue_name,
project=project_id,
client_uuid=client_uuid,
**kwargs)
# Buffer messages
cursor = next(results)
messages = list(cursor)
except validation.ValidationFailed as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPBadRequestAPI(str(ex))
except storage_errors.QueueDoesNotExist as ex:
LOG.debug(ex)
messages = None
except Exception:
description = _('Messages could not be listed.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
if not messages:
messages = []
else:
# Found some messages, so prepare the response
kwargs['marker'] = next(results)
base_path = req.path.rsplit('/', 1)[0]
messages = [wsgi_utils.format_message_v1_1(m, base_path,
m['claim_id'])
for m in messages]
links = []
if messages:
links = [
{
'rel': 'next',
'href': req.path + falcon.to_query_str(kwargs)
}
]
return {
'messages': messages,
'links': links
}
# ----------------------------------------------------------------------
# Interface
# ----------------------------------------------------------------------
@decorators.TransportLog("Messages collection")
def on_post(self, req, resp, project_id, queue_name):
client_uuid = wsgi_helpers.get_client_uuid(req)
try:
# Place JSON size restriction before parsing
self._validate.message_length(req.content_length)
except validation.ValidationFailed as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPBadRequestAPI(str(ex))
# Deserialize and validate the incoming messages
document = wsgi_utils.deserialize(req.stream, req.content_length)
if 'messages' not in document:
description = _('No messages were found in the request body.')
raise wsgi_errors.HTTPBadRequestAPI(description)
messages = wsgi_utils.sanitize(document['messages'],
self._message_post_spec,
doctype=wsgi_utils.JSONArray)
try:
self._validate.message_posting(messages)
if not self._queue_controller.exists(queue_name, project_id):
self._queue_controller.create(queue_name, project=project_id)
message_ids = self._message_controller.post(
queue_name,
messages=messages,
project=project_id,
client_uuid=client_uuid)
except validation.ValidationFailed as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPBadRequestAPI(str(ex))
except storage_errors.DoesNotExist as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPNotFound(str(ex))
except storage_errors.MessageConflict:
description = _('No messages could be enqueued.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
except Exception:
description = _('Messages could not be enqueued.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
# Prepare the response
ids_value = ','.join(message_ids)
resp.location = req.path + '?ids=' + ids_value
hrefs = [req.path + '/' + id for id in message_ids]
body = {'resources': hrefs}
resp.text = utils.to_json(body)
resp.status = falcon.HTTP_201
@decorators.TransportLog("Messages collection")
def on_get(self, req, resp, project_id, queue_name):
ids = req.get_param_as_list('ids')
if ids is None:
response = self._get(req, project_id, queue_name)
else:
response = self._get_by_id(req.path.rsplit('/', 1)[0], project_id,
queue_name, ids)
if response is None:
# NOTE(TheSriram): Trying to get a message by id, should
# return the message if its present, otherwise a 404 since
# the message might have been deleted.
msg = _('No messages with IDs: {ids} found in the queue {queue} '
'for project {project}.')
description = msg.format(queue=queue_name, project=project_id,
ids=ids)
raise wsgi_errors.HTTPNotFound(description)
else:
resp.text = utils.to_json(response)
# status defaults to 200
@decorators.TransportLog("Messages collection")
def on_delete(self, req, resp, project_id, queue_name):
ids = req.get_param_as_list('ids')
pop_limit = req.get_param_as_int('pop')
try:
self._validate.message_deletion(ids, pop_limit)
except validation.ValidationFailed as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPBadRequestAPI(str(ex))
if ids:
resp.status = self._delete_messages_by_id(queue_name, ids,
project_id)
elif pop_limit:
resp.status, resp.text = self._pop_messages(queue_name,
project_id,
pop_limit)
def _delete_messages_by_id(self, queue_name, ids, project_id):
try:
self._message_controller.bulk_delete(
queue_name,
message_ids=ids,
project=project_id)
except Exception:
description = _('Messages could not be deleted.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
return falcon.HTTP_204
def _pop_messages(self, queue_name, project_id, pop_limit):
try:
LOG.debug('POP messages - queue: %(queue)s, '
'project: %(project)s',
{'queue': queue_name, 'project': project_id})
messages = self._message_controller.pop(
queue_name,
project=project_id,
limit=pop_limit)
except Exception:
description = _('Messages could not be popped.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
# Prepare response
if not messages:
messages = []
body = {'messages': messages}
body = utils.to_json(body)
return falcon.HTTP_200, body
class ItemResource(object):
__slots__ = '_message_controller'
def __init__(self, message_controller):
self._message_controller = message_controller
@decorators.TransportLog("Messages item")
def on_get(self, req, resp, project_id, queue_name, message_id):
try:
message = self._message_controller.get(
queue_name,
message_id,
project=project_id)
except storage_errors.DoesNotExist as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPNotFound(str(ex))
except Exception:
description = _('Message could not be retrieved.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
# Prepare response
message['href'] = req.path
message = wsgi_utils.format_message_v1_1(message,
req.path.rsplit('/', 2)[0],
message['claim_id'])
resp.text = utils.to_json(message)
# status defaults to 200
@decorators.TransportLog("Messages item")
def on_delete(self, req, resp, project_id, queue_name, message_id):
error_title = _('Unable to delete')
try:
self._message_controller.delete(
queue_name,
message_id=message_id,
project=project_id,
claim=req.get_param('claim_id'))
except storage_errors.MessageNotClaimed as ex:
LOG.debug(ex)
description = _('A claim was specified, but the message '
'is not currently claimed.')
raise falcon.HTTPBadRequest(
title=error_title, description=description)
except storage_errors.ClaimDoesNotExist as ex:
LOG.debug(ex)
description = _('The specified claim does not exist or '
'has expired.')
raise falcon.HTTPBadRequest(
title=error_title, description=description)
except storage_errors.NotPermitted as ex:
LOG.debug(ex)
description = _('This message is claimed; it cannot be '
'deleted without a valid claim ID.')
raise falcon.HTTPForbidden(
title=error_title, description=description)
except Exception:
description = _('Message could not be deleted.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
# Alles guete
resp.status = falcon.HTTP_204
zaqar/transport/wsgi/v1_1/ping.py
-30
View File
@@ -1,30 +0,0 @@
# Copyright 2014 IBM Corp. All Rights Reserved.
#
# 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 falcon
class Resource(object):
__slots__ = ('_driver',)
def __init__(self, driver):
self._driver = driver
def on_get(self, req, resp, **kwargs):
resp.status = (falcon.HTTP_204 if self._driver.is_alive()
else falcon.HTTP_503)
def on_head(self, req, resp, **kwargs):
resp.status = falcon.HTTP_204
zaqar/transport/wsgi/v1_1/pools.py
-251
View File
@@ -1,251 +0,0 @@
# Copyright (c) 2013 Rackspace Hosting, 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.
"""pools: a resource to handle storage pool management
A pool is added by an operator by interacting with the
pooling-related endpoints. When specifying a pool, the
following fields are required:
::
{
"name": string,
"weight": integer,
"uri": string::uri
}
Furthermore, depending on the underlying storage type of pool being
registered, there is an optional field:
::
{
"options": {...}
}
"""
import falcon
import jsonschema
from oslo_log import log
from zaqar.common.api.schemas import pools as schema
from zaqar.common import utils as common_utils
from zaqar.i18n import _
from zaqar.storage import errors
from zaqar.storage import utils as storage_utils
from zaqar.transport import utils as transport_utils
from zaqar.transport.wsgi import errors as wsgi_errors
from zaqar.transport.wsgi import utils as wsgi_utils
LOG = log.getLogger(__name__)
class Listing(object):
"""A resource to list registered pools
:param pools_controller: means to interact with storage
"""
def __init__(self, pools_controller):
self._ctrl = pools_controller
def on_get(self, request, response, project_id):
"""Returns a pool listing as objects embedded in an object:
::
{
"pools": [
{"href": "", "weight": 100, "uri": ""},
...
],
"links": [
{"href": "", "rel": "next"}
]
}
:returns: HTTP | 200
"""
LOG.debug('LIST pools')
store = {}
request.get_param('marker', store=store)
request.get_param_as_int('limit', store=store)
request.get_param_as_bool('detailed', store=store)
cursor = self._ctrl.list(**store)
pools = list(next(cursor))
results = {'links': []}
if pools:
store['marker'] = next(cursor)
for entry in pools:
entry['href'] = request.path + '/' + entry['name']
results['links'] = [
{
'rel': 'next',
'href': request.path + falcon.to_query_str(store)
}
]
results['pools'] = pools
response.content_location = request.relative_uri
response.text = transport_utils.to_json(results)
response.status = falcon.HTTP_200
class Resource(object):
"""A handler for individual pool.
:param pools_controller: means to interact with storage
"""
def __init__(self, pools_controller):
self._ctrl = pools_controller
validator_type = jsonschema.Draft4Validator
self._validators = {
'weight': validator_type(schema.patch_weight),
'uri': validator_type(schema.patch_uri),
'group': validator_type(schema.patch_uri),
'options': validator_type(schema.patch_options),
'create': validator_type(schema.create)
}
def on_get(self, request, response, project_id, pool):
"""Returns a JSON object for a single pool entry:
::
{"weight": 100, "uri": "", options: {...}}
:returns: HTTP | [200, 404]
"""
LOG.debug('GET pool - name: %s', pool)
data = None
detailed = request.get_param_as_bool('detailed') or False
try:
data = self._ctrl.get(pool, detailed)
except errors.PoolDoesNotExist as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPNotFound(str(ex))
data['href'] = request.path
response.text = transport_utils.to_json(data)
def on_put(self, request, response, project_id, pool):
"""Registers a new pool. Expects the following input:
::
{"weight": 100, "uri": ""}
An options object may also be provided.
:returns: HTTP | [201, 204]
"""
LOG.debug('PUT pool - name: %s', pool)
conf = self._ctrl.driver.conf
data = wsgi_utils.load(request)
wsgi_utils.validate(self._validators['create'], data)
if not storage_utils.can_connect(data['uri'], conf=conf):
raise wsgi_errors.HTTPBadRequestBody(
'cannot connect to %s' % data['uri']
)
try:
self._ctrl.create(pool, weight=data['weight'],
uri=data['uri'],
options=data.get('options', {}))
response.status = falcon.HTTP_201
response.location = request.path
except errors.PoolCapabilitiesMismatch as e:
title = _('Unable to create pool')
LOG.exception(title)
raise falcon.HTTPBadRequest(title=title, description=str(e))
except errors.PoolAlreadyExists as e:
LOG.exception('Pool "%s" already exists', pool)
raise wsgi_errors.HTTPConflict(str(e))
def on_delete(self, request, response, project_id, pool):
"""Deregisters a pool.
:returns: HTTP | [204, 403]
"""
LOG.debug('DELETE pool - name: %s', pool)
try:
self._ctrl.delete(pool)
except errors.PoolInUseByFlavor as ex:
title = _('Unable to delete')
description = _('This pool is used by flavors {flavor}; '
'It cannot be deleted.')
description = description.format(flavor=ex.flavor)
LOG.exception(description)
raise falcon.HTTPForbidden(title=title, description=description)
response.status = falcon.HTTP_204
def on_patch(self, request, response, project_id, pool):
"""Allows one to update a pool's weight, uri, and/or options.
This method expects the user to submit a JSON object
containing at least one of: 'uri', 'weight', 'group', 'options'. If
none are found, the request is flagged as bad. There is also
strict format checking through the use of
jsonschema. Appropriate errors are returned in each case for
badly formatted input.
:returns: HTTP | 200,400
"""
LOG.debug('PATCH pool - name: %s', pool)
data = wsgi_utils.load(request)
EXPECT = ('weight', 'uri', 'options')
if not any([(field in data) for field in EXPECT]):
LOG.debug('PATCH pool, bad params')
raise wsgi_errors.HTTPBadRequestBody(
'One of `uri`, `weight`,or `options` needs '
'to be specified'
)
for field in EXPECT:
wsgi_utils.validate(self._validators[field], data)
conf = self._ctrl.driver.conf
if 'uri' in data and not storage_utils.can_connect(data['uri'],
conf=conf):
raise wsgi_errors.HTTPBadRequestBody(
'cannot connect to %s' % data['uri']
)
fields = common_utils.fields(data, EXPECT,
pred=lambda v: v is not None)
try:
self._ctrl.update(pool, **fields)
except errors.PoolDoesNotExist as ex:
LOG.exception('Pool "%s" does not exist', pool)
raise wsgi_errors.HTTPNotFound(str(ex))
zaqar/transport/wsgi/v1_1/queues.py
-161
View File
@@ -1,161 +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 falcon
from oslo_log import log as logging
from zaqar.common import decorators
from zaqar.i18n import _
from zaqar.storage import errors as storage_errors
from zaqar.transport import utils
from zaqar.transport import validation
from zaqar.transport.wsgi import errors as wsgi_errors
from zaqar.transport.wsgi import utils as wsgi_utils
LOG = logging.getLogger(__name__)
class ItemResource(object):
__slots__ = ('_validate', '_queue_controller', '_message_controller')
def __init__(self, validate, queue_controller, message_controller):
self._validate = validate
self._queue_controller = queue_controller
self._message_controller = message_controller
@decorators.TransportLog("Queue metadata")
def on_get(self, req, resp, project_id, queue_name):
try:
resp_dict = self._queue_controller.get(queue_name,
project=project_id)
except storage_errors.DoesNotExist as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPNotFound(str(ex))
except Exception:
description = _('Queue metadata could not be retrieved.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
resp.text = utils.to_json(resp_dict)
# status defaults to 200
@decorators.TransportLog("Queue item")
def on_put(self, req, resp, project_id, queue_name):
try:
# Place JSON size restriction before parsing
self._validate.queue_metadata_length(req.content_length)
# Deserialize queue metadata
metadata = None
if req.content_length:
document = wsgi_utils.deserialize(req.stream,
req.content_length)
metadata = wsgi_utils.sanitize(document)
# NOTE(Eva-i): reserved queue attributes is Zaqar's feature since
# API v2. But we have to ensure the bad data will not come from
# older APIs, so we validate metadata here.
self._validate.queue_metadata_putting(metadata)
except validation.ValidationFailed as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPBadRequestAPI(str(ex))
try:
created = self._queue_controller.create(queue_name,
metadata=metadata,
project=project_id)
except storage_errors.FlavorDoesNotExist as ex:
LOG.exception('"%s" does not exist', queue_name)
raise wsgi_errors.HTTPBadRequestAPI(str(ex))
except Exception:
description = _('Queue could not be created.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
resp.status = falcon.HTTP_201 if created else falcon.HTTP_204
resp.location = req.path
@decorators.TransportLog("Queue item")
def on_delete(self, req, resp, project_id, queue_name):
try:
self._queue_controller.delete(queue_name, project=project_id)
except Exception:
description = _('Queue could not be deleted.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
resp.status = falcon.HTTP_204
class CollectionResource(object):
__slots__ = ('_queue_controller', '_validate')
def __init__(self, validate, queue_controller):
self._queue_controller = queue_controller
self._validate = validate
@decorators.TransportLog("Queue collection")
def on_get(self, req, resp, project_id):
kwargs = {}
# NOTE(kgriffs): This syntax ensures that
# we don't clobber default values with None.
req.get_param('marker', store=kwargs)
req.get_param_as_int('limit', store=kwargs)
req.get_param_as_bool('detailed', store=kwargs)
try:
self._validate.queue_listing(**kwargs)
results = self._queue_controller.list(project=project_id, **kwargs)
# Buffer list of queues
queues = list(next(results))
except validation.ValidationFailed as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPBadRequestAPI(str(ex))
except Exception:
description = _('Queues could not be listed.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
# Got some. Prepare the response.
kwargs['marker'] = next(results) or kwargs.get('marker', '')
for each_queue in queues:
each_queue['href'] = req.path + '/' + each_queue['name']
links = []
if queues:
links = [
{
'rel': 'next',
'href': req.path + falcon.to_query_str(kwargs)
}
]
response_body = {
'queues': queues,
'links': links
}
resp.text = utils.to_json(response_body)
# status defaults to 200
zaqar/transport/wsgi/v1_1/stats.py
-73
View File
@@ -1,73 +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.
from oslo_log import log as logging
from zaqar.i18n import _
from zaqar.storage import errors as storage_errors
from zaqar.transport import utils
from zaqar.transport.wsgi import errors as wsgi_errors
LOG = logging.getLogger(__name__)
class Resource(object):
__slots__ = '_queue_ctrl'
def __init__(self, queue_controller):
self._queue_ctrl = queue_controller
def on_get(self, req, resp, project_id, queue_name):
try:
resp_dict = self._queue_ctrl.stats(queue_name,
project=project_id)
message_stats = resp_dict['messages']
if message_stats['total'] != 0:
base_path = req.path[:req.path.rindex('/')] + '/messages/'
newest = message_stats['newest']
newest['href'] = base_path + newest['id']
del newest['id']
oldest = message_stats['oldest']
oldest['href'] = base_path + oldest['id']
del oldest['id']
resp.text = utils.to_json(resp_dict)
# status defaults to 200
except (storage_errors.QueueDoesNotExist,
storage_errors.QueueIsEmpty):
resp_dict = {
'messages': {
'claimed': 0,
'free': 0,
'total': 0
}
}
resp.text = utils.to_json(resp_dict)
except storage_errors.DoesNotExist as ex:
LOG.debug(ex)
raise wsgi_errors.HTTPNotFound(str(ex))
except Exception:
description = _('Queue stats could not be read.')
LOG.exception(description)
raise wsgi_errors.HTTPServiceUnavailable(description)
zaqar/transport/wsgi/version.py
-2
View File
@@ -15,12 +15,10 @@
import falcon
from zaqar.transport import utils
from zaqar.transport.wsgi import v1_1
from zaqar.transport.wsgi import v2_0
VERSIONS = {
'versions': [
v1_1.VERSION,
v2_0.VERSION
]
}