TrueCloudLab/s3-tests
7
0
Fork 1
mirror of https://github.com/ceph/s3-tests.git synced 2025-02-17 05:47:39 +00:00
Code Issues 2 Projects Releases Packages Wiki Activity

s3: clean up use of tenanted bucket names

Browse source 
the botocore.handlers.validate_bucket_name validation is enabled by
default, but we can disable it with unregister(). this avoids having to
mess with the signature or url

Signed-off-by: Casey Bodley <cbodley@redhat.com>
This commit is contained in:
Casey Bodley 2024-09-06 08:34:44 -04:00
parent aa82bd16ae
commit aad858679c
1 changed files with 5 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
s3tests_boto3/functional/test_s3.py
View file

@ -2,6 +2,7 @@ import boto3
import botocore.session
from botocore.exceptions import ClientError
from botocore.exceptions import ParamValidationError
from botocore.handlers import validate_bucket_name
import isodate
import email.utils
import datetime
@ -10574,20 +10575,6 @@ def test_bucketv2_policy_acl():
client.delete_bucket_policy(Bucket=bucket_name)
client.put_bucket_acl(Bucket=bucket_name, ACL='public-read')
def tenanted_bucket_name(tenant):
def change_bucket_name(params, **kwargs):
old_name = params['context']['signing']['bucket']
new_name = "{}:{}".format(tenant, old_name)
params['Bucket'] = new_name
params['context']['signing']['bucket'] = new_name
# the : needs to be url-encoded for urls
new_name_url = "{}%3A{}".format(tenant, old_name)
params['url'] = params['url'].replace(old_name, new_name_url)
params['url_path'] = params['url_path'].replace(old_name, new_name_url)
return change_bucket_name
@pytest.mark.bucket_policy
def test_bucket_policy_different_tenant():
bucket_name = get_new_bucket()
@ -10615,8 +10602,8 @@ def test_bucket_policy_different_tenant():
# use the tenanted client to list the global tenant's bucket
tenant_client = get_tenant_client()
tenant_client.meta.events.register('before-call.s3.ListObjects', tenanted_bucket_name(''))
response = tenant_client.list_objects(Bucket=bucket_name)
tenant_client.meta.events.unregister("before-parameter-build.s3", validate_bucket_name)
response = tenant_client.list_objects(Bucket=":{}".format(bucket_name))
assert len(response['Contents']) == 1
@ -10649,9 +10636,9 @@ def test_bucket_policy_tenanted_bucket():
# use the global tenant's client to list the tenanted bucket
client = get_client()
client.meta.events.register('before-call.s3.ListObjects', tenanted_bucket_name(tenant))
client.meta.events.unregister("before-parameter-build.s3", validate_bucket_name)
response = client.list_objects(Bucket=bucket_name)
response = client.list_objects(Bucket="{}:{}".format(tenant, bucket_name))
assert len(response['Contents']) == 1
@pytest.mark.bucket_policy