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

Add test to verify HTTP OPTIONS on presigned URL

Browse source 
Related: https://tracker.ceph.com/issues/62033

Signed-off-by: Tobias Urdin <tobias.urdin@binero.com>
(cherry picked from commit c0a1880d4c)

Conflicts:
	s3tests_boto3/functional/test_s3.py: nose->pytest changes
(cherry picked from commit 8fb5d9a59c52adda9cc8172c96611398944be478)
This commit is contained in:
Tobias Urdin 2023-08-02 09:28:20 +00:00 committed by Casey Bodley
parent e2aa00d394
commit bbf65028e5
1 changed files with 51 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

60
s3tests_boto3/functional/test_s3.py
View file

@ -3696,13 +3696,14 @@ def test_put_object_ifnonmatch_overwrite_existed_failed():
body = _get_body(response) body = _get_body(response)
eq(body, 'bar') eq(body, 'bar')
def _setup_bucket_object_acl(bucket_acl, object_acl): def _setup_bucket_object_acl(bucket_acl, object_acl, client=None):
""" """
add a foo key, and specified key and bucket acls to add a foo key, and specified key and bucket acls to
a (new or existing) bucket. a (new or existing) bucket.
""" """
bucket_name = get_new_bucket_name() if client is None:
client = get_client() client = get_client()
bucket_name = get_new_bucket_name()
client.create_bucket(ACL=bucket_acl, Bucket=bucket_name) client.create_bucket(ACL=bucket_acl, Bucket=bucket_name)
client.put_object(ACL=object_acl, Bucket=bucket_name, Key='foo') client.put_object(ACL=object_acl, Bucket=bucket_name, Key='foo')
@ -3934,19 +3935,31 @@ def test_object_raw_authenticated_object_gone():
eq(status, 404) eq(status, 404)
eq(error_code, 'NoSuchKey') eq(error_code, 'NoSuchKey')
def _test_object_raw_get_x_amz_expires_not_expired(client):
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')
res = requests.options(url, verify=get_config_ssl_verify()).__dict__
eq(res['status_code'], 400)
res = requests.get(url, verify=get_config_ssl_verify()).__dict__
eq(res['status_code'], 200)
@attr(resource='object') @attr(resource='object')
@attr(method='get') @attr(method='get')
@attr(operation='x-amz-expires check not expired') @attr(operation='x-amz-expires check not expired')
@attr(assertion='succeeds') @attr(assertion='succeeds')
def test_object_raw_get_x_amz_expires_not_expired(): def test_object_raw_get_x_amz_expires_not_expired():
bucket_name = _setup_bucket_object_acl('public-read', 'public-read') _test_object_raw_get_x_amz_expires_not_expired(client=get_client())
client = get_client()
params = {'Bucket': bucket_name, 'Key': 'foo'}
url = client.generate_presigned_url(ClientMethod='get_object', Params=params, ExpiresIn=100000, HttpMethod='GET') @attr(resource='object')
@attr(method='get')
res = requests.get(url, verify=get_config_ssl_verify()).__dict__ @attr(operation='x-amz-expires check not expired')
eq(res['status_code'], 200) @attr(assertion='succeeds')
def test_object_raw_get_x_amz_expires_not_expired_tenant():
_test_object_raw_get_x_amz_expires_not_expired(client=get_tenant_client())
@attr(resource='object') @attr(resource='object')
@attr(method='get') @attr(method='get')
@ -7550,6 +7563,35 @@ 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) _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):
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')
res = requests.options(url, verify=get_config_ssl_verify()).__dict__
eq(res['status_code'], 400)
allowed_methods = ['GET']
allowed_origins = ['example']
cors_config ={
'CORSRules': [
{'AllowedMethods': allowed_methods,
'AllowedOrigins': allowed_origins,
},
]
}
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')
def test_cors_presigned_get_object():
_test_cors_options_presigned_get_object(client=get_client())
def test_cors_presigned_get_object_tenant():
_test_cors_options_presigned_get_object(client=get_tenant_client())
@attr(resource='bucket') @attr(resource='bucket')
@attr(method='put') @attr(method='put')
@attr(operation='put tags') @attr(operation='put tags')