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

Merge pull request #260 from cbodley/wip-boto3-ssl

Browse source 
Wip boto3 ssl
This commit is contained in:
Ali Maredia 2019-03-07 12:12:00 -05:00 committed by GitHub
commit 5be61bc30a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 20 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

50
s3tests_boto3/functional/__init__.py
View file

@ -155,7 +155,10 @@ def setup():
# vars from the DEFAULT section
config.default_host = defaults.get("host")
config.default_port = int(defaults.get("port"))
config.default_is_secure = defaults.get("is_secure")
config.default_is_secure = cfg.getboolean('DEFAULT', "is_secure")
proto = 'https' if config.default_is_secure else 'http'
config.default_endpoint = "%s://%s:%d" % (proto, config.default_host, config.default_port)
# vars from the main section
config.main_access_key = cfg.get('s3 main',"access_key")
@ -211,27 +214,20 @@ def get_client(client_config=None):
if client_config == None:
client_config = Config(signature_version='s3v4')
endpoint_url = "http://%s:%d" % (config.default_host, config.default_port)
client = boto3.client(service_name='s3',
aws_access_key_id=config.main_access_key,
aws_secret_access_key=config.main_secret_key,
endpoint_url=endpoint_url,
endpoint_url=config.default_endpoint,
use_ssl=config.default_is_secure,
verify=False,
config=client_config)
return client
def get_v2_client():
endpoint_url = "http://%s:%d" % (config.default_host, config.default_port)
client = boto3.client(service_name='s3',
aws_access_key_id=config.main_access_key,
aws_secret_access_key=config.main_secret_key,
endpoint_url=endpoint_url,
endpoint_url=config.default_endpoint,
use_ssl=config.default_is_secure,
verify=False,
config=Config(signature_version='s3'))
return client
@ -239,14 +235,11 @@ def get_alt_client(client_config=None):
if client_config == None:
client_config = Config(signature_version='s3v4')
endpoint_url = "http://%s:%d" % (config.default_host, config.default_port)
client = boto3.client(service_name='s3',
aws_access_key_id=config.alt_access_key,
aws_secret_access_key=config.alt_secret_key,
endpoint_url=endpoint_url,
endpoint_url=config.default_endpoint,
use_ssl=config.default_is_secure,
verify=False,
config=client_config)
return client
@ -254,40 +247,29 @@ def get_tenant_client(client_config=None):
if client_config == None:
client_config = Config(signature_version='s3v4')
endpoint_url = "http://%s:%d" % (config.default_host, config.default_port)
client = boto3.client(service_name='s3',
aws_access_key_id=config.tenant_access_key,
aws_secret_access_key=config.tenant_secret_key,
endpoint_url=endpoint_url,
endpoint_url=config.default_endpoint,
use_ssl=config.default_is_secure,
verify=False,
config=client_config)
return client
def get_unauthenticated_client():
endpoint_url = "http://%s:%d" % (config.default_host, config.default_port)
client = boto3.client(service_name='s3',
aws_access_key_id='',
aws_secret_access_key='',
endpoint_url=endpoint_url,
endpoint_url=config.default_endpoint,
use_ssl=config.default_is_secure,
verify=False,
config=Config(signature_version=UNSIGNED))
return client
def get_bad_auth_client(aws_access_key_id='badauth'):
endpoint_url = "http://%s:%d" % (config.default_host, config.default_port)
client = boto3.client(service_name='s3',
aws_access_key_id=aws_access_key_id,
aws_secret_access_key='roflmao',
endpoint_url=endpoint_url,
endpoint_url=config.default_endpoint,
use_ssl=config.default_is_secure,
verify=False,
config=Config(signature_version='s3v4'))
return client
@ -314,14 +296,11 @@ def get_new_bucket_resource(name=None):
Always recreates a bucket from scratch. This is useful to also
reset ACLs and such.
"""
endpoint_url = "http://%s:%d" % (config.default_host, config.default_port)
s3 = boto3.resource('s3',
use_ssl=False,
verify=False,
endpoint_url=endpoint_url,
aws_access_key_id=config.main_access_key,
aws_secret_access_key=config.main_secret_key)
aws_secret_access_key=config.main_secret_key,
endpoint_url=config.default_endpoint,
use_ssl=config.default_is_secure)
if name is None:
name = get_new_bucket_name()
bucket = s3.Bucket(name)
@ -353,6 +332,9 @@ def get_config_host():
def get_config_port():
return config.default_port
def get_config_endpoint():
return config.default_endpoint
def get_main_aws_access_key():
return config.main_access_key

17
s3tests_boto3/functional/test_s3.py
View file

@ -50,6 +50,7 @@ from . import (
get_config_is_secure,
get_config_host,
get_config_port,
get_config_endpoint,
get_main_aws_access_key,
get_main_aws_secret_key,
get_main_display_name,
@ -1360,18 +1361,8 @@ def test_object_write_file():
eq(body, 'bar')
def _get_post_url(bucket_name):
protocol='http'
is_secure = get_config_is_secure()
if is_secure is True:
protocol='https'
host = get_config_host()
port = get_config_port()
url = '{protocol}://{host}:{port}/{bucket_name}'.format(protocol=protocol,\
host=host, port=port, bucket_name=bucket_name)
return url
endpoint = get_config_endpoint()
return '{endpoint}/{bucket_name}'.format(endpoint=endpoint, bucket_name=bucket_name)
@attr(resource='object')
@attr(method='post')
@ -6189,7 +6180,7 @@ def _simple_http_req_100_cont(host, port, is_secure, method, resource):
)
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if(is_secure == True):
if is_secure:
s = ssl.wrap_socket(s);
s.settimeout(5)
s.connect((host, port))