TrueCloudLab/s3-tests
7
0
Fork 1
mirror of https://github.com/ceph/s3-tests.git synced 2025-02-21 15:28:06 +00:00
Code Issues 2 Projects Releases Packages Wiki Activity

Fix for #493

Browse source 
Changes from pull - https://github.com/ceph/s3-tests/pull/490
This commit is contained in:
Vishnu Vardhan 2023-02-18 20:57:44 -08:00
parent b1472019d7
commit 9a8863785f
1 changed files with 21 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
s3tests_boto3/functional/__init__.py
View file

@ -82,18 +82,13 @@ def get_objects_list(bucket, client=None, prefix=None):
# generator function that returns object listings in batches, where each # generator function that returns object listings in batches, where each
# batch is a list of dicts compatible with delete_objects() # batch is a list of dicts compatible with delete_objects()
def list_versions(client, bucket, batch_size): def list_versions(client, bucket, batch_size):
key_marker = '' kwargs = {'Bucket': bucket, 'MaxKeys': batch_size}
version_marker = ''
truncated = True truncated = True
while truncated: while truncated:
listing = client.list_object_versions( listing = client.list_object_versions(**kwargs)
Bucket=bucket,
KeyMarker=key_marker,
VersionIdMarker=version_marker,
MaxKeys=batch_size)
key_marker = listing.get('NextKeyMarker') kwargs['KeyMarker'] = listing.get('NextKeyMarker')
version_marker = listing.get('NextVersionIdMarker') kwargs['VersionIdMarker'] = listing.get('NextVersionIdMarker')
truncated = listing['IsTruncated'] truncated = listing['IsTruncated']
objs = listing.get('Versions', []) + listing.get('DeleteMarkers', []) objs = listing.get('Versions', []) + listing.get('DeleteMarkers', [])
@ -104,11 +99,25 @@ def nuke_bucket(client, bucket):
batch_size = 128 batch_size = 128
max_retain_date = None max_retain_date = None
try:
response = client.get_object_lock_configuration(Bucket=bucket)
LockEnabled = True
except ClientError as error:
if error.response['Error']['Code'] == 'ObjectLockConfigurationNotFoundError':
LockEnabled = False
else:
raise error
# list and delete objects in batches # list and delete objects in batches
for objects in list_versions(client, bucket, batch_size): for objects in list_versions(client, bucket, batch_size):
if(LockEnabled):
delete = client.delete_objects(Bucket=bucket, delete = client.delete_objects(Bucket=bucket,
Delete={'Objects': objects, 'Quiet': True}, Delete={'Objects': objects, 'Quiet': True},
BypassGovernanceRetention=True) BypassGovernanceRetention = LockEnabled)
else:
delete = client.delete_objects(Bucket=bucket,
Delete={'Objects': objects, 'Quiet': True})
# check for object locks on 403 AccessDenied errors # check for object locks on 403 AccessDenied errors
for err in delete.get('Errors', []): for err in delete.get('Errors', []):