TrueCloudLab/s3-tests
7
0
Fork 1
mirror of https://github.com/ceph/s3-tests.git synced 2024-11-21 23:29:47 +00:00
Code Issues 1 Projects Releases Packages Wiki Activity

refactor out _make_request and _make_bucket_request, with a common _make_raw_request; ready for s3website testing.

Browse source 
Signed-off-by: Robin H. Johnson <robbat2@gentoo.org>
This commit is contained in:
Robin H. Johnson 2015-06-15 05:18:07 +00:00 committed by Robin H. Johnson
parent 3cf86161d0
commit a48a8983b7
2 changed files with 89 additions and 54 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

87
s3tests/functional/__init__.py
View file

@ -8,6 +8,8 @@ import itertools
import os
import random
import string
from httplib import HTTPConnection, HTTPSConnection
from urlparse import urlparse
from .utils import region_sync_meta
@ -312,6 +314,10 @@ def setup():
'user_id',
'display_name',
'email',
's3website_domain',
'host',
'port',
'is_secure',
]:
try:
config[name][var] = cfg.get(section, var)
@ -394,3 +400,84 @@ def get_new_bucket(target=None, name=None, headers=None):
# ignore that as astronomically unlikely
bucket = connection.create_bucket(name, location=target.conf.api_name, headers=headers)
return bucket
def _make_request(method, bucket, key, body=None, authenticated=False, response_headers=None, request_headers=None, expires_in=100000, path_style=True):
"""
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)
"""
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 = key.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)
else:
path = '/{obj}'.format(bucket=key.bucket.name, obj=key.name)
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)
def _make_bucket_request(method, bucket, body=None, authenticated=False, response_headers=None, request_headers=None, expires_in=100000, path_style=True):
"""
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)
"""
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)
def _make_raw_request(host, port, method, path, body=None, request_headers=None, secure=False):
if secure:
class_ = HTTPSConnection
else:
class_ = HTTPConnection
if request_headers is None:
request_headers = {}
skip_host=('Host' in request_headers)
skip_accept_encoding = False
c = class_(host, port, strict=True)
# We do the request manually, so we can muck with headers
#c.request(method, path, body=body, headers=request_headers)
c.connect()
c.putrequest(method, path, skip_host, skip_accept_encoding)
for k,v in request_headers.items():
c.putheader(k,v)
c.endheaders(message_body=body)
res = c.getresponse()
#c.close()
print(res.status, res.reason)
return res

56
s3tests/functional/test_s3.py
View file

@ -27,9 +27,6 @@ import re
import xml.etree.ElementTree as ET
from httplib import HTTPConnection, HTTPSConnection
from urlparse import urlparse
from nose.tools import eq_ as eq
from nose.plugins.attrib import attr
from nose.plugins.skip import SkipTest
@ -53,6 +50,8 @@ from . import (
config,
get_prefix,
is_slow_backend,
_make_request,
_make_bucket_request,
)
@ -2455,57 +2454,6 @@ def _setup_bucket_request(bucket_acl=None):
return bucket
def _make_request(method, bucket, key, body=None, authenticated=False, response_headers=None, expires_in=100000):
"""
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)
"""
if authenticated:
url = key.generate_url(expires_in, method=method, response_headers=response_headers)
o = urlparse(url)
path = o.path + '?' + o.query
else:
path = '/{bucket}/{obj}'.format(bucket=key.bucket.name, obj=key.name)
if s3.main.is_secure:
class_ = HTTPSConnection
else:
class_ = HTTPConnection
c = class_(s3.main.host, s3.main.port, strict=True)
c.request(method, path, body=body)
res = c.getresponse()
print res.status, res.reason
return res
def _make_bucket_request(method, bucket, body=None, authenticated=False, expires_in=100000):
"""
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)
"""
if authenticated:
url = bucket.generate_url(expires_in, method=method)
o = urlparse(url)
path = o.path + '?' + o.query
else:
path = '/{bucket}'.format(bucket=bucket.name)
if s3.main.is_secure:
class_ = HTTPSConnection
else:
class_ = HTTPConnection
c = class_(s3.main.host, s3.main.port, strict=True)
c.request(method, path, body=body)
res = c.getresponse()
print res.status, res.reason
return res
@attr(resource='object')
@attr(method='get')
@attr(operation='publically readable bucket')