dkirillov/s3-tests
1
0
Fork 0
forked from TrueCloudLab/s3-tests
Code Pull requests Activity

Refactor _make_bucket_request.

Browse source 
s3tests.functional._make_bucket_request is now a special case of
_make_request, as requested by
https://github.com/ceph/s3-tests/pull/92#commitcomment-15775972

Signed-off-by: Robin H. Johnson <robin.johnson@dreamhost.com>
This commit is contained in:
Robin H. Johnson 2016-01-30 17:38:45 +00:00
parent f028e1e926
commit 94fa405270
1 changed files with 31 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
s3tests/functional/__init__.py
View file

@ -405,7 +405,9 @@ def _make_request(method, bucket, key, body=None, authenticated=False, response_
"""
issue a request for a specified method, on a specified <bucket,key>,
with a specified (optional) body (encrypted per the connection), and
return the response (status, reason)
return the response (status, reason).
If key is None, then this will be treated as a bucket-level request.
"""
if response_headers is None:
response_headers = {}
@ -416,44 +418,48 @@ def _make_request(method, bucket, key, body=None, authenticated=False, response_
request_headers['Host'] = conn.calling_format.build_host(conn.server_name(), bucket.name)
if authenticated:
url = key.generate_url(expires_in, method=method, response_headers=response_headers, headers=request_headers)
urlobj = None
if key is not None:
urlobj = key
elif bucket is not None:
urlobj = bucket
else:
raise RuntimeError('Unable to find bucket name')
url = urlobj.generate_url(expires_in, method=method, response_headers=response_headers, headers=request_headers)
o = urlparse(url)
path = o.path + '?' + o.query
else:
if path_style:
path = '/{bucket}/{obj}'.format(bucket=key.bucket.name, obj=key.name)
bucketobj = None
if key is not None:
path = '/{obj}'.format(obj=key.name)
bucketobj = key.bucket
elif bucket is not None:
path = '/'
bucketobj = bucket
else:
path = '/{obj}'.format(bucket=key.bucket.name, obj=key.name)
raise RuntimeError('Unable to find bucket name')
if path_style:
path = '/{bucket}'.format(bucket=bucketobj.name) + path
return _make_raw_request(host=s3.main.host, port=s3.main.port, method=method, path=path, body=body, request_headers=request_headers, secure=s3.main.is_secure, timeout=timeout)
def _make_bucket_request(method, bucket, body=None, authenticated=False, response_headers=None, request_headers=None, expires_in=100000, path_style=True, timeout=None):
"""
issue a request for a specified method, on a specified <bucket,key>,
issue a request for a specified method, on a specified <bucket>,
with a specified (optional) body (encrypted per the connection), and
return the response (status, reason)
"""
if response_headers is None:
response_headers = {}
if request_headers is None:
request_headers = {}
if not path_style:
conn = bucket.connection
request_headers['Host'] = conn.calling_format.build_host(conn.server_name(), bucket.name)
if authenticated:
url = bucket.generate_url(expires_in, method=method, response_headers=response_headers, headers=request_headers)
o = urlparse(url)
path = o.path + '?' + o.query
else:
if path_style:
path = '/{bucket}'.format(bucket=bucket.name)
else:
path = '/'
return _make_raw_request(host=s3.main.host, port=s3.main.port, method=method, path=path, body=body, request_headers=request_headers, secure=s3.main.is_secure, timeout=timeout)
return _make_request(method=method, bucket=bucket, key=None, body=body, authenticated=authenticated, response_headers=response_headers, request_headers=request_headers, expires_in=expires_in, path_style=path_style, timeout=timeout)
def _make_raw_request(host, port, method, path, body=None, request_headers=None, secure=False, timeout=None):
"""
issue a request to a specific host & port, for a specified method, on a
specified path with a specified (optional) body (encrypted per the
connection), and return the response (status, reason).
This allows construction of special cases not covered by the bucket/key to
URL mapping of _make_request/_make_bucket_request.
"""
if secure:
class_ = HTTPSConnection
else: