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

Merge pull request #544 from tobias-urdin/improve-presigned

Browse source 
Add more testing for presigned URLs
This commit is contained in:
Casey Bodley 2024-02-12 19:51:43 +00:00 committed by GitHub
commit 1d5764d569
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 84 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

90
s3tests_boto3/functional/test_s3.py
View file

@ -3522,6 +3522,36 @@ def test_object_put_authenticated():
response = client.put_object(Bucket=bucket_name, Key='foo', Body='foo')
assert response['ResponseMetadata']['HTTPStatusCode'] == 200
def _test_object_presigned_put_object_with_acl(client=None):
if client is None:
client = get_client()
bucket_name = get_new_bucket(client)
key = 'foo'
params = {'Bucket': bucket_name, 'Key': key, 'ACL': 'private'}
url = client.generate_presigned_url(ClientMethod='put_object', Params=params, HttpMethod='PUT')
data = b'hello world'
headers = {'x-amz-acl': 'private'}
res = requests.put(url, data=data, headers=headers, verify=get_config_ssl_verify())
assert res.status_code == 200
params = {'Bucket': bucket_name, 'Key': key}
url = client.generate_presigned_url(ClientMethod='get_object', Params=params, HttpMethod='GET')
res = requests.get(url, verify=get_config_ssl_verify())
assert res.status_code == 200
assert res.text == 'hello world'
def test_object_presigned_put_object_with_acl():
_test_object_presigned_put_object_with_acl(
client=get_client())
def test_object_presigned_put_object_with_acl_tenant():
_test_object_presigned_put_object_with_acl(
client=get_tenant_client())
def test_object_raw_put_authenticated_expired():
bucket_name = get_new_bucket()
client = get_client()
@ -6692,16 +6722,26 @@ def test_cors_header_option():
_cors_request_and_check(requests.options, obj_url, {'Origin': 'example.origin','Access-Control-Request-Headers':'x-amz-meta-header2','Access-Control-Request-Method':'GET'}, 403, None, None)
def _test_cors_options_presigned_get_object(client):
def _test_cors_options_presigned_method(client, method, cannedACL=None):
bucket_name = _setup_bucket_object_acl('public-read', 'public-read', client=client)
params = {'Bucket': bucket_name, 'Key': 'foo'}
url = client.generate_presigned_url(ClientMethod='get_object', Params=params, ExpiresIn=100000, HttpMethod='GET')
if cannedACL is not None:
params['ACL'] = cannedACL
if method == 'get_object':
httpMethod = 'GET'
elif method == 'put_object':
httpMethod = 'PUT'
else:
raise ValueError('invalid method')
url = client.generate_presigned_url(ClientMethod=method, Params=params, ExpiresIn=100000, HttpMethod=httpMethod)
res = requests.options(url, verify=get_config_ssl_verify()).__dict__
assert res['status_code'] == 400
allowed_methods = ['GET']
allowed_methods = [httpMethod]
allowed_origins = ['example']
cors_config ={
@ -6713,13 +6753,51 @@ def _test_cors_options_presigned_get_object(client):
}
client.put_bucket_cors(Bucket=bucket_name, CORSConfiguration=cors_config)
_cors_request_and_check(requests.options, url, {'Origin': 'example', 'Access-Control-Request-Method': 'GET'}, 200, 'example', 'GET')
headers = {
'Origin': 'example',
'Access-Control-Request-Method': httpMethod,
}
_cors_request_and_check(requests.options, url, headers,
200, 'example', httpMethod)
def test_cors_presigned_get_object():
_test_cors_options_presigned_get_object(client=get_client())
_test_cors_options_presigned_method(
client=get_client(),
method='get_object',
)
def test_cors_presigned_get_object_tenant():
_test_cors_options_presigned_get_object(client=get_tenant_client())
_test_cors_options_presigned_method(
client=get_tenant_client(),
method='get_object',
)
def test_cors_presigned_put_object():
_test_cors_options_presigned_method(
client=get_client(),
method='put_object',
)
def test_cors_presigned_put_object_with_acl():
_test_cors_options_presigned_method(
client=get_client(),
method='put_object',
cannedACL='private',
)
def test_cors_presigned_put_object_tenant():
_test_cors_options_presigned_method(
client=get_tenant_client(),
method='put_object',
)
def test_cors_presigned_put_object_tenant_with_acl():
_test_cors_options_presigned_method(
client=get_tenant_client(),
method='put_object',
cannedACL='private',
)
@pytest.mark.tagging
def test_set_bucket_tagging():