openstack/zaqar
Code Issues Proposed changes

Merge "Support to query single resource with pre-signed url"

Browse Source
This commit is contained in:
Jenkins 2017-07-10 02:01:14 +00:00 committed by Gerrit Code Review
commit 2c6654a0ca
2 changed files with 59 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
zaqar/common/transport/wsgi/helpers.py
View File

@ -16,6 +16,7 @@
"""wsgi transport helpers.""" """wsgi transport helpers."""
from distutils import version from distutils import version
import re
import uuid import uuid
import falcon import falcon
@ -45,7 +46,8 @@ def verify_pre_signed_url(key, req, resp, params):
if req.method not in methods: if req.method not in methods:
raise falcon.HTTPNotFound() raise falcon.HTTPNotFound()
if req.path not in paths: # Support to query single resource with pre-signed url
if not any([p for p in paths if re.search(p, req.path)]):
raise falcon.HTTPNotFound() raise falcon.HTTPNotFound()
try: try:

56
zaqar/tests/unit/transport/wsgi/v2_0/test_urls.py
View File

@ -110,6 +110,37 @@ class TestURL(base.V2Base):
response = self.simulate_get(content['paths'][0], headers=headers) response = self.simulate_get(content['paths'][0], headers=headers)
self.assertEqual(falcon.HTTP_200, self.srmock.status) self.assertEqual(falcon.HTTP_200, self.srmock.status)
def _get_msg_id(self, headers):
return self._get_msg_ids(headers)[0]
def _get_msg_ids(self, headers):
return headers['location'].rsplit('=', 1)[-1].split(',')
def test_url_verification_success_with_message_id(self):
doc = {'messages': [{'body': 239, 'ttl': 300}]}
body = jsonutils.dumps(doc)
self.simulate_post(self.url_prefix + '/queues/shared_queue/messages',
body=body, headers=self.headers)
msg_id = self._get_msg_id(self.srmock.headers_dict)
data = {'methods': ['GET', 'POST']}
response = self.simulate_post(self.signed_url_prefix,
body=jsonutils.dumps(data))
self.assertEqual(falcon.HTTP_200, self.srmock.status)
content = jsonutils.loads(response[0])
headers = {
'URL-Signature': content['signature'],
'URL-Expires': content['expires'],
'URL-Methods': ','.join(content['methods']),
'URL-Paths': ','.join(content['paths'])
}
headers.update(self.headers)
self.simulate_get(content['paths'][0] + '/' + msg_id,
headers=headers)
self.assertEqual(falcon.HTTP_200, self.srmock.status)
def test_url_verification_bad_request(self): def test_url_verification_bad_request(self):
path = self.url_prefix + '/queues/shared_queue/messages' path = self.url_prefix + '/queues/shared_queue/messages'
expires = timeutils.utcnow() + datetime.timedelta(days=1) expires = timeutils.utcnow() + datetime.timedelta(days=1)
@ -174,3 +205,28 @@ class TestURL(base.V2Base):
headers.update(self.headers) headers.update(self.headers)
self.simulate_get(path, headers=headers) self.simulate_get(path, headers=headers)
self.assertEqual(falcon.HTTP_404, self.srmock.status) self.assertEqual(falcon.HTTP_404, self.srmock.status)
def test_url_verification_bad_with_message_id(self):
doc = {'messages': [{'body': 239, 'ttl': 300}]}
body = jsonutils.dumps(doc)
self.simulate_post(self.url_prefix + '/queues/shared_queue/messages',
body=body, headers=self.headers)
msg_id = self._get_msg_id(self.srmock.headers_dict)
data = {'methods': ['GET', 'POST']}
response = self.simulate_post(self.signed_url_prefix,
body=jsonutils.dumps(data))
self.assertEqual(falcon.HTTP_200, self.srmock.status)
content = jsonutils.loads(response[0])
headers = {
'URL-Signature': content['signature'],
'URL-Expires': content['expires'],
'URL-Methods': ','.join(content['methods']),
'URL-Paths': ','.join('/queues/shared_queue/claims')
}
headers.update(self.headers)
self.simulate_get(content['paths'][0] + '/' + msg_id,
headers=headers)
self.assertEqual(falcon.HTTP_404, self.srmock.status)