KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use list_bucket_inventory_configurations method in localstack

Best Python code snippet using localstack_python

test_bucket_inventory.py

Source:test_bucket_inventory.py

...104                    inventory_destination=InventoryDestination(bucket_destination=bucket_destination))105            self.bucket1.put_bucket_inventory_configuration(inventory_configuration);106        107        # test with param empty108        result = self.bucket1.list_bucket_inventory_configurations('')109        self.assertEquals(3, len(result.inventory_configurations))110        self.assertEquals(False, result.is_truncated)111        self.assertEquals(None, result.continuaiton_token)112        self.assertEquals(None, result.next_continuation_token)113        # test with param None114        result = self.bucket1.list_bucket_inventory_configurations()115        self.assertEquals(3, len(result.inventory_configurations))116        self.assertEquals(False, result.is_truncated)117        self.assertEquals(None, result.continuaiton_token)118        self.assertEquals(None, result.next_continuation_token)119        dest_bucket.delete_bucket()120    def test_list_lot_bucket_inventory(self):121        auth = oss2.Auth(OSS_ID, OSS_SECRET)122        dest_bucket_name = OSS_BUCKET + "-test-inventory-dest"123        dest_bucket = oss2.Bucket(auth, self.endpoint, dest_bucket_name)124        dest_bucket.create_bucket()125        inventory_id_prefix = "test-id-"126        for index in range(0, 120):127            inventory_id = inventory_id_prefix + str(index)128            optional_fields = [FIELD_SIZE, FIELD_LAST_MODIFIED_DATE, FIELD_STORAG_CLASS,129                    FIELD_ETAG, FIELD_IS_MULTIPART_UPLOADED, FIELD_ENCRYPTION_STATUS]130            bucket_destination = InventoryBucketDestination(131                    account_id=OSS_INVENTORY_BUCKET_DESTINATION_ACCOUNT, 132                    role_arn=OSS_INVENTORY_BUCKET_DESTINATION_ARN,133                    bucket=dest_bucket_name,134                    inventory_format=INVENTORY_FORMAT_CSV,135                    prefix="destination-prefix",136                    sse_kms_encryption=InventoryServerSideEncryptionKMS("test-kms-id"))137            inventory_configuration = InventoryConfiguration(138                    inventory_id=inventory_id,139                    is_enabled=True, 140                    inventory_schedule=InventorySchedule(frequency=INVENTORY_FREQUENCY_DAILY),141                    included_object_versions=INVENTORY_INCLUDED_OBJECT_VERSIONS_CURRENT,142                    inventory_filter=InventoryFilter(prefix="obj-prefix"),143                    optional_fields=optional_fields,144                    inventory_destination=InventoryDestination(bucket_destination=bucket_destination))145            self.bucket1.put_bucket_inventory_configuration(inventory_configuration);146        result = self.bucket1.list_bucket_inventory_configurations()147        self.assertEquals(100, len(result.inventory_configurations))148        self.assertEquals(True, result.is_truncated)149        self.assertEquals(None, result.continuaiton_token)150        self.assertIsNotNone(result.next_continuation_token)151        next_continuation_token = result.next_continuation_token152        result = self.bucket1.list_bucket_inventory_configurations(next_continuation_token)153        self.assertEquals(20, len(result.inventory_configurations))154        self.assertEquals(False, result.is_truncated)155        self.assertEquals(next_continuation_token, result.continuaiton_token)156        self.assertEquals(None, result.next_continuation_token)157        dest_bucket.delete_bucket()158    def test_list_none_inventory(self):159        self.assertRaises(oss2.exceptions.NoSuchInventory, self.bucket1.list_bucket_inventory_configurations)160if __name__ == '__main__':...

s3-objects-mass-encyption.py

Source:s3-objects-mass-encyption.py

...64    return string65def get_inventory_list(s3, bucket):66    no_inventory = False67    try:68        s3.list_bucket_inventory_configurations(Bucket=bucket)['InventoryConfigurationList']69    except:70        no_inventory = True71    if no_inventory:72        logger.error("Inventory list was NOT generated for the bucket " + bucket)73        return None74    else:75        invent_id     = s3.list_bucket_inventory_configurations(Bucket=bucket)['InventoryConfigurationList'][0]['Id']76        invent_bucket = s3.list_bucket_inventory_configurations(Bucket=bucket)['InventoryConfigurationList'][0]['Destination']['S3BucketDestination']['Bucket'].split(":")[-1]77        invent_format = s3.list_bucket_inventory_configurations(Bucket=bucket)['InventoryConfigurationList'][0]['Destination']['S3BucketDestination']['Format']78        if invent_format != 'CSV':79            logger.error("Unsupported format of the inventory list for the bucket '{}'. Only CSV format is currently supported.".format(bucket))80            return None81        invent_objs = s3.list_objects(Bucket=invent_bucket, Prefix="{}/{}/data/".format(bucket,invent_id))82        invent_date = 083        if len(invent_objs['Contents']) == 1:84            logger.warning("No inventory list found for the bucket '{}'! Either the bucket is empty or no inventory list generated yet.".format(bucket))85            return None86        for obj in invent_objs['Contents']:87            if obj['Key'][-1] == '/':88                continue89            if invent_date == 0:90                invent_date = obj['LastModified']91            if obj['LastModified'] >= invent_date:...

s3-bucket.py

Source:s3-bucket.py

...4    result = [bucket["Name"] for bucket in response["Buckets"]]5    return result6def print_inventory_config(client):7    for bucket in get_all_bucket(client):8        response = client.list_bucket_inventory_configurations(Bucket=bucket)9        config = response.get("InventoryConfigurationList", 0)10        if config and len(response["InventoryConfigurationList"]) > 1:11            print(12                f'bucket: {bucket} IncludedObjectVersions: {response["InventoryConfigurationList"]}'13            )14        elif config and len(response["InventoryConfigurationList"]) == 1:15            print(16                f'bucket: {bucket} IncludedObjectVersions: {response["InventoryConfigurationList"][0]["IncludedObjectVersions"]}'17            )18        else:19            print(f"bucket: {bucket} Inventory is not enabled")20def modify_inventory_config(client):21    config_bucket = {22        "us-east-1": "arn:aws:s3:::651732464523-inventory",23        "us-east-2": "arn:aws:s3:::651732464523-inventory-us-east-2",24        "us-west-2": "arn:aws:s3:::651732464523-inventory-us-west-2",25    }26    default_settings = {27        "Destination": {28            "S3BucketDestination": {29                "AccountId": "651732464523",30                "Bucket": "arn:aws:s3:::651732464523-inventory",31                "Format": "CSV",32            }33        },34        "IsEnabled": True,35        "Id": "daily-inventory-files",36        "IncludedObjectVersions": "Current",37        "OptionalFields": [38            "Size",39            "LastModifiedDate",40            "StorageClass",41            "ETag",42            "IsMultipartUploaded",43            "ReplicationStatus",44            "EncryptionStatus",45            "BucketKeyStatus",46            "IntelligentTieringAccessTier",47            "ObjectLockMode",48            "ObjectLockRetainUntilDate",49            "ObjectLockLegalHoldStatus",50        ],51        "Schedule": {"Frequency": "Daily"},52    }53    bucket_list = [54        "179268807624-us-east-1-migrated-179268807624",55        "1a27ks9b8uxpd-migrated-179268807624-2",56        "651732464523-build-state-bucket",57    ]58    for bucket in get_all_bucket(client):59        # for bucket in bucket_list:60        response = client.list_bucket_inventory_configurations(Bucket=bucket)61        config = response.get("InventoryConfigurationList", 0)62        current_region = "us-east-1"63        print(f"bucket: {bucket}")64        if config:65            if len(config) > 1:66                print(f"\tContains too many inventory config; manual review needed")67            else:68                print(f"\tModifying existing inventory configuration.")69                location_response = client.get_bucket_location(Bucket=bucket)70                new_location = (71                    location_response["LocationConstraint"]72                    if location_response["LocationConstraint"]73                    else "us-east-1"74                )...

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.