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

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>
This commit is contained in:
Tobias Urdin 2023-08-02 09:28:20 +00:00
parent 73b340a0e2
commit c0a1880d4c
1 changed files with 43 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
s3tests_boto3/functional/test_s3.py
View file

@ -3187,13 +3187,14 @@ def test_put_object_ifnonmatch_overwrite_existed_failed():
body = _get_body(response) body = _get_body(response)
assert body == 'bar' assert 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.
""" """
if client is None:
client = get_client()
bucket_name = get_new_bucket_name() bucket_name = get_new_bucket_name()
client = get_client()
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')
@ -3403,16 +3404,24 @@ def test_object_raw_authenticated_object_gone():
assert status == 404 assert status == 404
assert error_code == 'NoSuchKey' assert error_code == 'NoSuchKey'
def test_object_raw_get_x_amz_expires_not_expired(): def _test_object_raw_get_x_amz_expires_not_expired(client):
bucket_name = _setup_bucket_object_acl('public-read', 'public-read') bucket_name = _setup_bucket_object_acl('public-read', 'public-read', client=client)
client = get_client()
params = {'Bucket': bucket_name, 'Key': 'foo'} params = {'Bucket': bucket_name, 'Key': 'foo'}
url = client.generate_presigned_url(ClientMethod='get_object', Params=params, ExpiresIn=100000, HttpMethod='GET') url = client.generate_presigned_url(ClientMethod='get_object', Params=params, ExpiresIn=100000, HttpMethod='GET')
res = requests.options(url, verify=get_config_ssl_verify()).__dict__
assert res['status_code'] == 400
res = requests.get(url, verify=get_config_ssl_verify()).__dict__ res = requests.get(url, verify=get_config_ssl_verify()).__dict__
assert res['status_code'] == 200 assert res['status_code'] == 200
def test_object_raw_get_x_amz_expires_not_expired():
_test_object_raw_get_x_amz_expires_not_expired(client=get_client())
def test_object_raw_get_x_amz_expires_not_expired_tenant():
_test_object_raw_get_x_amz_expires_not_expired(client=get_tenant_client())
def test_object_raw_get_x_amz_expires_out_range_zero(): def test_object_raw_get_x_amz_expires_out_range_zero():
bucket_name = _setup_bucket_object_acl('public-read', 'public-read') bucket_name = _setup_bucket_object_acl('public-read', 'public-read')
client = get_client() client = get_client()
@ -6581,6 +6590,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__
assert 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())
@pytest.mark.tagging @pytest.mark.tagging
def test_set_bucket_tagging(): def test_set_bucket_tagging():
bucket_name = get_new_bucket() bucket_name = get_new_bucket()