How to use list_object_versions method in localstack

Best Python code snippet using localstack_python

test_versioning.py

Source:test_versioning.py Github

copy

Full Screen

...119 'Size': len(obj_data),120 'StorageClass': 'STANDARD',121 }], objs)122 # object version listing123 resp = self.client.list_object_versions(Bucket=self.bucket_name)124 objs = resp.get('Versions', [])125 for obj in objs:126 obj.pop('LastModified')127 obj.pop('Owner')128 obj.pop('VersionId')129 self.assertEqual([{130 'ETag': '"%s"' % obj_etag,131 'IsLatest': True,132 'Key': obj_name,133 'Size': len(obj_data),134 'StorageClass': 'STANDARD',135 }], objs)136 # overwrite the object137 new_obj_data = self.create_name('some-new-data').encode('ascii')138 new_obj_etag = hashlib.md5(new_obj_data).hexdigest()139 self.client.upload_fileobj(six.BytesIO(new_obj_data),140 self.bucket_name, obj_name)141 # new object is in the listing142 resp = self.client.list_objects_v2(Bucket=self.bucket_name)143 objs = resp.get('Contents', [])144 for obj in objs:145 obj.pop('LastModified')146 self.assertEqual([{147 'ETag': '"%s"' % new_obj_etag,148 'Key': obj_name,149 'Size': len(new_obj_data),150 'StorageClass': 'STANDARD',151 }], objs)152 # both object versions in the versions listing153 resp = self.client.list_object_versions(Bucket=self.bucket_name)154 objs = resp.get('Versions', [])155 for obj in objs:156 obj.pop('LastModified')157 obj.pop('Owner')158 obj.pop('VersionId')159 self.assertEqual([{160 'ETag': '"%s"' % new_obj_etag,161 'IsLatest': True,162 'Key': obj_name,163 'Size': len(new_obj_data),164 'StorageClass': 'STANDARD',165 }, {166 'ETag': '"%s"' % obj_etag,167 'IsLatest': False,168 'Key': obj_name,169 'Size': len(obj_data),170 'StorageClass': 'STANDARD',171 }], objs)172 def test_delete_versioned_objects(self):173 etags = []174 obj_name = self.create_name('versioned-obj')175 for i in range(3):176 obj_data = self.create_name('some-data-%s' % i).encode('ascii')177 etags.insert(0, hashlib.md5(obj_data).hexdigest())178 self.client.upload_fileobj(six.BytesIO(obj_data),179 self.bucket_name, obj_name)180 # only one object appears in the listing181 resp = self.client.list_objects_v2(Bucket=self.bucket_name)182 objs = resp.get('Contents', [])183 for obj in objs:184 obj.pop('LastModified')185 self.assertEqual([{186 'ETag': '"%s"' % etags[0],187 'Key': obj_name,188 'Size': len(obj_data),189 'StorageClass': 'STANDARD',190 }], objs)191 # but everything is layed out in the object versions listing192 resp = self.client.list_object_versions(Bucket=self.bucket_name)193 objs = resp.get('Versions', [])194 versions = []195 for obj in objs:196 obj.pop('LastModified')197 obj.pop('Owner')198 versions.append(obj.pop('VersionId'))199 self.assertEqual([{200 'ETag': '"%s"' % etags[0],201 'IsLatest': True,202 'Key': obj_name,203 'Size': len(obj_data),204 'StorageClass': 'STANDARD',205 }, {206 'ETag': '"%s"' % etags[1],207 'IsLatest': False,208 'Key': obj_name,209 'Size': len(obj_data),210 'StorageClass': 'STANDARD',211 }, {212 'ETag': '"%s"' % etags[2],213 'IsLatest': False,214 'Key': obj_name,215 'Size': len(obj_data),216 'StorageClass': 'STANDARD',217 }], objs)218 # we can delete a specific version219 resp = self.client.delete_object(Bucket=self.bucket_name,220 Key=obj_name,221 VersionId=versions[1])222 # and that just pulls it out of the versions listing223 resp = self.client.list_object_versions(Bucket=self.bucket_name)224 objs = resp.get('Versions', [])225 for obj in objs:226 obj.pop('LastModified')227 obj.pop('Owner')228 obj.pop('VersionId')229 self.assertEqual([{230 'ETag': '"%s"' % etags[0],231 'IsLatest': True,232 'Key': obj_name,233 'Size': len(obj_data),234 'StorageClass': 'STANDARD',235 }, {236 'ETag': '"%s"' % etags[2],237 'IsLatest': False,238 'Key': obj_name,239 'Size': len(obj_data),240 'StorageClass': 'STANDARD',241 }], objs)242 # ... but the current listing is unaffected243 resp = self.client.list_objects_v2(Bucket=self.bucket_name)244 objs = resp.get('Contents', [])245 for obj in objs:246 obj.pop('LastModified')247 self.assertEqual([{248 'ETag': '"%s"' % etags[0],249 'Key': obj_name,250 'Size': len(obj_data),251 'StorageClass': 'STANDARD',252 }], objs)253 # OTOH, if you delete specifically the latest version254 # we can delete a specific version255 resp = self.client.delete_object(Bucket=self.bucket_name,256 Key=obj_name,257 VersionId=versions[0])258 # the versions listing has a new IsLatest259 resp = self.client.list_object_versions(Bucket=self.bucket_name)260 objs = resp.get('Versions', [])261 for obj in objs:262 obj.pop('LastModified')263 obj.pop('Owner')264 obj.pop('VersionId')265 self.assertEqual([{266 'ETag': '"%s"' % etags[2],267 'IsLatest': True,268 'Key': obj_name,269 'Size': len(obj_data),270 'StorageClass': 'STANDARD',271 }], objs)272 # and the stack pops273 resp = self.client.list_objects_v2(Bucket=self.bucket_name)274 objs = resp.get('Contents', [])275 for obj in objs:276 obj.pop('LastModified')277 self.assertEqual([{278 'ETag': '"%s"' % etags[2],279 'Key': obj_name,280 'Size': len(obj_data),281 'StorageClass': 'STANDARD',282 }], objs)283 def test_delete_versioned_deletes(self):284 etags = []285 obj_name = self.create_name('versioned-obj')286 for i in range(3):287 obj_data = self.create_name('some-data-%s' % i).encode('ascii')288 etags.insert(0, hashlib.md5(obj_data).hexdigest())289 self.client.upload_fileobj(six.BytesIO(obj_data),290 self.bucket_name, obj_name)291 # and make a delete marker292 self.client.delete_object(Bucket=self.bucket_name, Key=obj_name)293 # current listing is empty294 resp = self.client.list_objects_v2(Bucket=self.bucket_name)295 objs = resp.get('Contents', [])296 self.assertEqual([], objs)297 # but everything is in layed out in the versions listing298 resp = self.client.list_object_versions(Bucket=self.bucket_name)299 objs = resp.get('Versions', [])300 versions = []301 for obj in objs:302 obj.pop('LastModified')303 obj.pop('Owner')304 versions.append(obj.pop('VersionId'))305 self.assertEqual([{306 'ETag': '"%s"' % etag,307 'IsLatest': False,308 'Key': obj_name,309 'Size': len(obj_data),310 'StorageClass': 'STANDARD',311 } for etag in etags], objs)312 # ... plus the delete markers313 delete_markers = resp.get('DeleteMarkers', [])314 marker_versions = []315 for marker in delete_markers:316 marker.pop('LastModified')317 marker.pop('Owner')318 marker_versions.append(marker.pop('VersionId'))319 self.assertEqual([{320 'Key': obj_name,321 'IsLatest': is_latest,322 } for is_latest in (True, False, False)], delete_markers)323 # delete an old delete markers324 resp = self.client.delete_object(Bucket=self.bucket_name,325 Key=obj_name,326 VersionId=marker_versions[2])327 # since IsLatest is still marker we'll raise NoSuchKey328 with self.assertRaises(ClientError) as caught:329 resp = self.client.get_object(Bucket=self.bucket_name,330 Key=obj_name)331 expected_err = 'An error occurred (NoSuchKey) when calling the ' \332 'GetObject operation: The specified key does not exist.'333 self.assertEqual(expected_err, str(caught.exception))334 # now delete the delete marker (IsLatest)335 resp = self.client.delete_object(Bucket=self.bucket_name,336 Key=obj_name,337 VersionId=marker_versions[0])338 # most recent version is now latest339 resp = self.client.get_object(Bucket=self.bucket_name,340 Key=obj_name)341 self.assertEqual(200, resp['ResponseMetadata']['HTTPStatusCode'])342 self.assertEqual('"%s"' % etags[0], resp['ETag'])343 # now delete the IsLatest object version344 resp = self.client.delete_object(Bucket=self.bucket_name,345 Key=obj_name,346 VersionId=versions[0])347 # and object is deleted again348 with self.assertRaises(ClientError) as caught:349 resp = self.client.get_object(Bucket=self.bucket_name,350 Key=obj_name)351 expected_err = 'An error occurred (NoSuchKey) when calling the ' \352 'GetObject operation: The specified key does not exist.'353 self.assertEqual(expected_err, str(caught.exception))354 # delete marker IsLatest355 resp = self.client.list_object_versions(Bucket=self.bucket_name)356 delete_markers = resp.get('DeleteMarkers', [])357 for marker in delete_markers:358 marker.pop('LastModified')359 marker.pop('Owner')360 self.assertEqual([{361 'Key': obj_name,362 'IsLatest': True,363 'VersionId': marker_versions[1],364 }], delete_markers)365 def test_multipart_upload(self):366 obj_name = self.create_name('versioned-obj')367 obj_data = b'data'368 mu = self.client.create_multipart_upload(369 Bucket=self.bucket_name,370 Key=obj_name)371 part_md5 = self.client.upload_part(372 Bucket=self.bucket_name,373 Key=obj_name,374 UploadId=mu['UploadId'],375 PartNumber=1,376 Body=obj_data)['ETag']377 complete_response = self.client.complete_multipart_upload(378 Bucket=self.bucket_name,379 Key=obj_name,380 UploadId=mu['UploadId'],381 MultipartUpload={'Parts': [382 {'PartNumber': 1, 'ETag': part_md5},383 ]})384 obj_etag = complete_response['ETag']385 delete_response = self.client.delete_object(386 Bucket=self.bucket_name,387 Key=obj_name)388 marker_version_id = delete_response['VersionId']389 resp = self.client.list_object_versions(Bucket=self.bucket_name)390 objs = resp.get('Versions', [])391 versions = []392 for obj in objs:393 obj.pop('LastModified')394 obj.pop('Owner')395 versions.append(obj.pop('VersionId'))396 self.assertEqual([{397 'ETag': obj_etag,398 'IsLatest': False,399 'Key': obj_name,400 'Size': len(obj_data),401 'StorageClass': 'STANDARD',402 }], objs)403 markers = resp.get('DeleteMarkers', [])404 for marker in markers:405 marker.pop('LastModified')406 marker.pop('Owner')407 self.assertEqual([{408 'IsLatest': True,409 'Key': obj_name,410 'VersionId': marker_version_id,411 }], markers)412 # Can still get the old version413 resp = self.client.get_object(414 Bucket=self.bucket_name,415 Key=obj_name,416 VersionId=versions[0])417 self.assertEqual(200, resp['ResponseMetadata']['HTTPStatusCode'])418 self.assertEqual(obj_etag, resp['ETag'])419 delete_response = self.client.delete_object(420 Bucket=self.bucket_name,421 Key=obj_name,422 VersionId=versions[0])423 resp = self.client.list_object_versions(Bucket=self.bucket_name)424 self.assertEqual([], resp.get('Versions', []))425 markers = resp.get('DeleteMarkers', [])426 for marker in markers:427 marker.pop('LastModified')428 marker.pop('Owner')429 self.assertEqual([{430 'IsLatest': True,431 'Key': obj_name,432 'VersionId': marker_version_id,433 }], markers)434 def test_get_versioned_object(self):435 etags = []436 obj_name = self.create_name('versioned-obj')437 for i in range(3):438 obj_data = self.create_name('some-data-%s' % i).encode('ascii')439 # TODO: pull etag from response instead440 etags.insert(0, hashlib.md5(obj_data).hexdigest())441 self.client.upload_fileobj(442 six.BytesIO(obj_data), self.bucket_name, obj_name)443 resp = self.client.list_object_versions(Bucket=self.bucket_name)444 objs = resp.get('Versions', [])445 versions = []446 for obj in objs:447 obj.pop('LastModified')448 obj.pop('Owner')449 versions.append(obj.pop('VersionId'))450 self.assertEqual([{451 'ETag': '"%s"' % etags[0],452 'IsLatest': True,453 'Key': obj_name,454 'Size': len(obj_data),455 'StorageClass': 'STANDARD',456 }, {457 'ETag': '"%s"' % etags[1],458 'IsLatest': False,459 'Key': obj_name,460 'Size': len(obj_data),461 'StorageClass': 'STANDARD',462 }, {463 'ETag': '"%s"' % etags[2],464 'IsLatest': False,465 'Key': obj_name,466 'Size': len(obj_data),467 'StorageClass': 'STANDARD',468 }], objs)469 # un-versioned get_object returns IsLatest470 resp = self.client.get_object(Bucket=self.bucket_name, Key=obj_name)471 self.assertEqual(200, resp['ResponseMetadata']['HTTPStatusCode'])472 self.assertEqual('"%s"' % etags[0], resp['ETag'])473 # but you can get any object by version474 for i, version in enumerate(versions):475 resp = self.client.get_object(476 Bucket=self.bucket_name, Key=obj_name, VersionId=version)477 self.assertEqual(200, resp['ResponseMetadata']['HTTPStatusCode'])478 self.assertEqual('"%s"' % etags[i], resp['ETag'])479 # and head_object works about the same480 resp = self.client.head_object(Bucket=self.bucket_name, Key=obj_name)481 self.assertEqual(200, resp['ResponseMetadata']['HTTPStatusCode'])482 self.assertEqual('"%s"' % etags[0], resp['ETag'])483 self.assertEqual(versions[0], resp['VersionId'])484 for version, etag in zip(versions, etags):485 resp = self.client.head_object(486 Bucket=self.bucket_name, Key=obj_name, VersionId=version)487 self.assertEqual(200, resp['ResponseMetadata']['HTTPStatusCode'])488 self.assertEqual(version, resp['VersionId'])489 self.assertEqual('"%s"' % etag, resp['ETag'])490 def test_get_versioned_object_invalid_params(self):491 with self.assertRaises(ClientError) as ctx:492 self.client.list_object_versions(Bucket=self.bucket_name,493 KeyMarker='',494 VersionIdMarker='bogus')495 expected_err = 'An error occurred (InvalidArgument) when calling ' \496 'the ListObjectVersions operation: Invalid version id specified'497 self.assertEqual(expected_err, str(ctx.exception))498 with self.assertRaises(ClientError) as ctx:499 self.client.list_object_versions(500 Bucket=self.bucket_name,501 VersionIdMarker='a' * 32)502 expected_err = 'An error occurred (InvalidArgument) when calling ' \503 'the ListObjectVersions operation: A version-id marker cannot ' \504 'be specified without a key marker.'505 self.assertEqual(expected_err, str(ctx.exception))506 def test_get_versioned_object_key_marker(self):507 obj00_name = self.create_name('00-versioned-obj')508 obj01_name = self.create_name('01-versioned-obj')509 names = [obj00_name] * 3 + [obj01_name] * 3510 latest = [True, False, False, True, False, False]511 etags = []512 for i in range(3):513 obj_data = self.create_name('some-data-%s' % i).encode('ascii')514 etags.insert(0, '"%s"' % hashlib.md5(obj_data).hexdigest())515 self.client.upload_fileobj(516 six.BytesIO(obj_data), self.bucket_name, obj01_name)517 for i in range(3):518 obj_data = self.create_name('some-data-%s' % i).encode('ascii')519 etags.insert(0, '"%s"' % hashlib.md5(obj_data).hexdigest())520 self.client.upload_fileobj(521 six.BytesIO(obj_data), self.bucket_name, obj00_name)522 resp = self.client.list_object_versions(Bucket=self.bucket_name)523 versions = []524 objs = []525 for o in resp.get('Versions', []):526 versions.append(o['VersionId'])527 objs.append({528 'Key': o['Key'],529 'VersionId': o['VersionId'],530 'IsLatest': o['IsLatest'],531 'ETag': o['ETag'],532 })533 expected = [{534 'Key': name,535 'VersionId': version,536 'IsLatest': is_latest,537 'ETag': etag,538 } for name, etag, version, is_latest in zip(539 names, etags, versions, latest)]540 self.assertEqual(expected, objs)541 # on s3 this makes expected[0]['IsLatest'] magicaly change to False?542 # resp = self.client.list_object_versions(Bucket=self.bucket_name,543 # KeyMarker='',544 # VersionIdMarker=versions[0])545 # objs = [{546 # 'Key': o['Key'],547 # 'VersionId': o['VersionId'],548 # 'IsLatest': o['IsLatest'],549 # 'ETag': o['ETag'],550 # } for o in resp.get('Versions', [])]551 # self.assertEqual(expected, objs)552 # KeyMarker skips past that key553 resp = self.client.list_object_versions(Bucket=self.bucket_name,554 KeyMarker=obj00_name)555 objs = [{556 'Key': o['Key'],557 'VersionId': o['VersionId'],558 'IsLatest': o['IsLatest'],559 'ETag': o['ETag'],560 } for o in resp.get('Versions', [])]561 self.assertEqual(expected[3:], objs)562 # KeyMarker with VersionIdMarker skips past that version563 resp = self.client.list_object_versions(Bucket=self.bucket_name,564 KeyMarker=obj00_name,565 VersionIdMarker=versions[0])566 objs = [{567 'Key': o['Key'],568 'VersionId': o['VersionId'],569 'IsLatest': o['IsLatest'],570 'ETag': o['ETag'],571 } for o in resp.get('Versions', [])]572 self.assertEqual(expected[1:], objs)573 # KeyMarker with bogus version skips past that key574 resp = self.client.list_object_versions(575 Bucket=self.bucket_name,576 KeyMarker=obj00_name,577 VersionIdMarker=versions[4])578 objs = [{579 'Key': o['Key'],580 'VersionId': o['VersionId'],581 'IsLatest': o['IsLatest'],582 'ETag': o['ETag'],583 } for o in resp.get('Versions', [])]584 self.assertEqual(expected[3:], objs)585 def test_list_objects(self):586 etags = defaultdict(list)587 for i in range(3):588 obj_name = self.create_name('versioned-obj')589 for i in range(3):590 obj_data = self.create_name('some-data-%s' % i).encode('ascii')591 etags[obj_name].insert(0, hashlib.md5(obj_data).hexdigest())592 self.client.upload_fileobj(593 six.BytesIO(obj_data), self.bucket_name, obj_name)594 # both unversioned list_objects responses are similar595 expected = []596 for name, obj_etags in sorted(etags.items()):597 expected.append({598 'ETag': '"%s"' % obj_etags[0],599 'Key': name,600 'Size': len(obj_data),601 'StorageClass': 'STANDARD',602 })603 resp = self.client.list_objects(Bucket=self.bucket_name)604 objs = resp.get('Contents', [])605 for obj in objs:606 obj.pop('LastModified')607 # one difference seems to be the Owner key608 self.assertEqual({'DisplayName', 'ID'},609 set(obj.pop('Owner').keys()))610 self.assertEqual(expected, objs)611 resp = self.client.list_objects_v2(Bucket=self.bucket_name)612 objs = resp.get('Contents', [])613 for obj in objs:614 obj.pop('LastModified')615 self.assertEqual(expected, objs)616 # versioned listings has something for everyone617 expected = []618 for name, obj_etags in sorted(etags.items()):619 is_latest = True620 for etag in obj_etags:621 expected.append({622 'ETag': '"%s"' % etag,623 'IsLatest': is_latest,624 'Key': name,625 'Size': len(obj_data),626 'StorageClass': 'STANDARD',627 })628 is_latest = False629 resp = self.client.list_object_versions(Bucket=self.bucket_name)630 objs = resp.get('Versions', [])631 versions = []632 for obj in objs:633 obj.pop('LastModified')634 obj.pop('Owner')635 versions.append(obj.pop('VersionId'))636 self.assertEqual(expected, objs)637 def test_copy_object(self):638 etags = []639 obj_name = self.create_name('versioned-obj')640 for i in range(3):641 obj_data = self.create_name('some-data-%s' % i).encode('ascii')642 etags.insert(0, hashlib.md5(obj_data).hexdigest())643 self.client.upload_fileobj(644 six.BytesIO(obj_data), self.bucket_name, obj_name)645 resp = self.client.list_object_versions(Bucket=self.bucket_name)646 objs = resp.get('Versions', [])647 versions = []648 for obj in objs:649 versions.append(obj.pop('VersionId'))650 # CopySource can just be Bucket/Key string651 first_target = self.create_name('target-obj1')652 copy_resp = self.client.copy_object(653 Bucket=self.bucket_name, Key=first_target,654 CopySource='%s/%s' % (self.bucket_name, obj_name))655 self.assertEqual(versions[0], copy_resp['CopySourceVersionId'])656 # and you'll just get the most recent version657 resp = self.client.head_object(Bucket=self.bucket_name,658 Key=first_target)659 self.assertEqual(200, resp['ResponseMetadata']['HTTPStatusCode'])...

Full Screen

Full Screen

lambda_function.py

Source:lambda_function.py Github

copy

Full Screen

...83 84 if eventName == 'PutObject':85 fileName = event['detail']['requestParameters']['key']86 fileListModified.append(fileName)87 versionHistory = s3.list_object_versions(88 Bucket=bucketName,89 Prefix=fileName90 )91 versionId = ''92 for ver in versionHistory['Versions']:93 if ver['IsLatest'] == True:94 versionId = ver['VersionId']95 restoreOld = s3.delete_object(96 Bucket=bucketName, 97 Key =fileName, 98 VersionId = versionId 99 )100 101 if eventName == 'DeleteObjects':102 versionHistory = s3.list_object_versions(Bucket=bucketName)103 eventTime = event['detail']['eventTime']104 formatEventTime = datetime.strptime(eventTime, '%Y-%m-%dT%H:%M:%SZ')105 formatEventTime = formatEventTime.replace(tzinfo=tzutc())106 versionHistory = s3.list_object_versions(Bucket=bucketName)107 versionIdList = []108 # fileListModified = []109 for ver in versionHistory['DeleteMarkers']:110 if ver['LastModified'] > formatEventTime:111 versionIdList.append(ver['VersionId'])112 fileListModified.append(ver['Key'])113 for i in range(len(versionIdList)):114 restoreOld = s3.delete_object(115 Bucket=bucketName, 116 Key =fileListModified[i], 117 VersionId = versionIdList[i] 118 )119 120 response = sns.publish(121 TopicArn=sns_topic_arn, 122 Message= (username + ' performed an unauthorised ' + 123 eventName +' operation to ' + bucketName + 124 ' Bucket at time ' + eventTime + 125 ' from user agent '+ hostAgent + 126 ' with IP: '+ hostIPAddress + 127 " and modified files: " + ','.join(fileListModified)),128 Subject= bucketName + ' Bucket Modified', 129 MessageStructure='string'130 )131 else:132 print("Access already denied")133 134 if eventName == 'PutObject':135 eventTime = event['detail']['eventTime']136 formatEventTime = datetime.strptime(eventTime, '%Y-%m-%dT%H:%M:%SZ')137 formatEventTime = formatEventTime.replace(tzinfo=tzutc())138 fileName = event['detail']['requestParameters']['key']139 fileListModified.append(fileName)140 versionHistory = s3.list_object_versions(141 Bucket=bucketName,142 Prefix=fileName143 )144 145 versionId = ''146 for ver in versionHistory['Versions']:147 if ver['IsLatest'] == True and ver['LastModified'] > formatEventTime:148 versionId = ver['VersionId']149 restoreOld = s3.delete_object(150 Bucket=bucketName, 151 Key =fileName, 152 VersionId = versionId 153 )154 ...

Full Screen

Full Screen

utils.py

Source:utils.py Github

copy

Full Screen

...9 else:10 return client.list_objects_v2(Bucket=bucket, MaxKeys=1000, Prefix=prefix)11def _get_objects_batch_all_versions(client, bucket, marker):12 if marker:13 return client.list_object_versions(Bucket=bucket, MaxKeys=1000, KeyMarker=marker)14 else:15 return client.list_object_versions(Bucket=bucket, MaxKeys=1000)16def _get_objects_batch_all_versions_with_prefix(client, bucket, prefix, marker):17 if marker:18 return client.list_object_versions(Bucket=bucket, MaxKeys=1000, Prefix=prefix, KeyMarker=marker)19 else:20 return client.list_object_versions(Bucket=bucket, MaxKeys=1000, Prefix=prefix)21def do_for_all_objects_all_versions(client, bucket, prefix, func):22 has_more = True23 key_marker = None24 while has_more:25 if prefix:26 object_list = _get_objects_batch_all_versions_with_prefix(client, bucket, prefix, key_marker)27 else:28 object_list = _get_objects_batch_all_versions(client, bucket, key_marker)29 if 'Versions' in object_list:30 entries = object_list['Versions']31 for e in entries:32 func({'VersionId': e['VersionId'], 'Key': e['Key']})33 has_more = object_list['IsTruncated']34 key_marker = object_list['NextKeyMarker'] if has_more else None...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run localstack automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful