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

How to use s3_resource method in localstack

Best Python code snippet using localstack_python

test_index.py

Source:test_index.py

1# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.2# SPDX-License-Identifier: Apache-2.03from unittest import TestCase4from unittest.mock import patch, MagicMock5from botocore.exceptions import ClientError6with patch("boto3.resource") as boto_resource_mock:7    with patch("boto3.client") as boto_client_mock:8        with patch("gamekitresourcemanagement.cfn_custom_resource.send_success_response") as mock_send_success_response:9            with patch('gamekitresourcemanagement.cfn_custom_resource.send_failure_response') as mock_send_failure_response:10                from functions.main.EmptyS3BucketOnDelete import index11BUCKET_NAME = 'test_bucket'12LOG_GROUP_NAME = 'test_log_group'13PHYSICAL_RESOURCE_ID = 'empty-on-delete-some-hash-here'14STACK_ID = 'test_stack_id'15LOGICAL_RESOURCE_ID = 'test_logical_resource_id'16REQUEST_ID = 'test_request_id'17RESPONSE_URL = 'https://website.tld/some_url'18HEAD_OBJECT = 'HeadObject'19class TestIndex(TestCase):20    def setUp(self):21        index.s3_resource = MagicMock()22        index.s3_client = MagicMock()23        mock_send_success_response.reset_mock()24        mock_send_failure_response.reset_mock()25    def test_calls_delete_all_bucket_non_versioned_success(self):26        # Arrange27        event = self.get_event()28        context = self.get_context()29        self.validate_empties_non_versioned_bucket(BUCKET_NAME, event, context)30        index.s3_resource.Bucket.return_value.delete.assert_not_called()31    def test_calls_delete_all_bucket_versioned_success(self):32        # Arrange33        event = self.get_event()34        context = self.get_context()35        index.s3_resource.BucketVersioning.return_value.status = 'Enabled'36        index.s3_resource.Bucket.return_value.object_versions.all.return_value.delete.return_value = {'Deleted': []}37        # Act38        index.lambda_handler(event, context)39        # Assert40        index.s3_resource.Bucket.assert_called_with(BUCKET_NAME)41        index.s3_resource.Bucket.return_value.objects.all.assert_not_called()42        index.s3_resource.Bucket.return_value.object_versions.all.return_value.delete.assert_called_once()43        mock_send_success_response.assert_called_once()44        mock_send_failure_response.assert_not_called()45    def test_ignores_non_delete_events(self):46        # Arrange47        event = self.get_event()48        event['RequestType'] = 'Create'49        context = self.get_context()50        # Act51        index.lambda_handler(event, context)52        # Assert53        index.s3_resource.Bucket.assert_not_called()54        mock_send_success_response.assert_called_once()55        mock_send_failure_response.assert_not_called()56    def test_fails_on_head_bucket_forbidden(self):57        # Arrange58        event = self.get_event()59        context = self.get_context()60        index.s3_client.head_bucket.side_effect = ClientError({61            'Error': {62                'Code': '403',63                'Message': 'Forbidden'64            }65        }, HEAD_OBJECT)66        # Act67        index.lambda_handler(event, context)68        # Assert69        index.s3_resource.Bucket.assert_not_called()70        mock_send_success_response.assert_not_called()71        mock_send_failure_response.assert_called_once()72    def test_ignores_non_existent_bucket(self):73        # Arrange74        event = self.get_event()75        context = self.get_context()76        index.s3_client.head_bucket.side_effect = ClientError({77            'Error': {78                'Code': '404',79                'Message': 'Not Found'80            }81        }, HEAD_OBJECT)82        # Act83        index.lambda_handler(event, context)84        # Assert85        index.s3_resource.Bucket.assert_not_called()86        mock_send_success_response.assert_called_once()87        mock_send_failure_response.assert_not_called()88    def test_calls_delete_all_bucket_non_versioned_failed(self):89        # Arrange90        event = self.get_event()91        context = self.get_context()92        index.s3_resource.BucketVersioning.return_value.status = None93        index.s3_resource.Bucket.return_value.objects.all.side_effect = ClientError({94            'Error': {95                'Code': 500,96                'Message': 'Some S3 Error'97            }98        }, 'ListObjects')99        # Act100        index.lambda_handler(event, context)101        # Assert102        index.s3_resource.Bucket.assert_called_with(BUCKET_NAME)103        index.s3_resource.Bucket.return_value.objects.all.assert_called_once()104        index.s3_resource.Bucket.return_value.object_versions.all.assert_not_called()105        mock_send_success_response.assert_not_called()106        mock_send_failure_response.assert_called_once()107    def test_calls_delete_flag_set_true_success(self):108        # Arrange109        event = self.get_event()110        event['ResourceProperties']['delete_bucket'] = 'true'111        context = self.get_context()112        self.validate_empties_non_versioned_bucket(BUCKET_NAME, event, context)113        index.s3_resource.Bucket.return_value.delete.assert_called()114    def test_calls_delete_flag_set_false_success(self):115        # Arrange116        event = self.get_event()117        event['ResourceProperties']['delete_bucket'] = 'false'118        context = self.get_context()119        self.validate_empties_non_versioned_bucket(BUCKET_NAME, event, context)120        index.s3_resource.Bucket.return_value.delete.assert_not_called()121    @staticmethod122    def validate_empties_non_versioned_bucket(bucket_name, event, context):123        index.s3_resource.BucketVersioning.return_value.status = None124        index.s3_resource.Bucket.return_value.objects.all.return_value.delete.return_value = {'Deleted': []}125        # Act126        index.lambda_handler(event, context)127        # Assert128        index.s3_resource.Bucket.assert_called_with(bucket_name)129        index.s3_resource.Bucket.return_value.objects.all.return_value.delete.assert_called_once()130        index.s3_resource.Bucket.return_value.object_versions.all.assert_not_called()131        mock_send_success_response.assert_called_once()132        mock_send_failure_response.assert_not_called()133    @staticmethod134    def get_event():135        return {136            'RequestType': 'Delete',137            'ResourceProperties': {138                'bucket_name': BUCKET_NAME139            },140            'PhysicalResourceId': PHYSICAL_RESOURCE_ID,141            'StackId': STACK_ID,142            'LogicalResourceId': LOGICAL_RESOURCE_ID,143            'RequestId': REQUEST_ID,144            'ResponseURL': RESPONSE_URL145        }146    @staticmethod147    def get_context():148        context = MagicMock()149        context.log_group_name = LOG_GROUP_NAME150        return context151    @staticmethod152    def get_urllib_context_manager(status_code: int):153        context_manager = MagicMock()154        context_manager.getcode.return_value = status_code155        context_manager.reason.return_value = status_code156        context_manager.read.return_value = 'request_body'157        context_manager.__enter__.return_value = context_manager...

s3.py

Source:s3.py

1import uuid2import boto33s3_client = boto3.client('s3')4s3_resource = boto3.resource('s3')5def create_bucket_name(bucket_prefix):6    # random generate bucket name using uuid7    return ''.join([bucket_prefix, str(uuid.uuid4())])8def create_bucket(bucket_prefix, s3_connection):9    session = boto3.session.Session()10    current_region = session.region_name11    bucket_name = create_bucket_name(bucket_prefix)12    bucket_response = s3_connection.create_bucket(13        Bucket=bucket_name,14        CreateBucketConfiguration={15            'LocationConstraint': current_region})16    print(bucket_name, current_region)17    return bucket_name, bucket_response18def create_temp_file(size, file_name, file_content):19    random_file_name = ''.join([str(uuid.uuid4().hex[:6]), file_name])20    with open(random_file_name, 'w') as f:21        f.write(str(file_content) * size)22    return random_file_name23# copy objects between buckets24def copy_to_bucket(bucket_from_name, bucket_to_name, file_name):25    copy_source = {26        'Bucket': bucket_from_name,27        'Key': file_name28    }29    s3_resource.Object(bucket_to_name, file_name).copy(copy_source)30# delete objects (Key and VersionID need to be Capitalised)31def delete_all_objects(bucket_name):32    res = []33    bucket = s3_resource.Bucket(bucket_name)34    for obj_version in bucket.object_versions.all():35        res.append({'Key': obj_version.object_key,36                    'VersionId': obj_version.id})37    print(res)38    bucket.delete_objects(Delete={'Objects': res})39def main():40    '''41    # Create Bucket42    first_bucket_name, first_response = create_bucket(43        bucket_prefix='firstboto',44        s3_connection=s3_resource.meta.client)45    first_file_name = create_temp_file(300, 'firstfile.txt', 'f')46    print(first_file_name)47    first_bucket = s3_resource.Bucket(name=first_bucket_name)48    print(first_bucket)49    first_object = s3_resource.Object(50        bucket_name=first_bucket_name, key=first_file_name51    )52    print(first_object)53    # uploading file to an object or to a bucket54    first_object.upload_file(first_file_name)55    # print(s3_resource.Object(bucket_name=first_bucket_name,key=first_bucket_name).download_file(f'C:\\ge.yang\Desktop'))56    # Buckets and Objects Traversal57    for bucket in s3_resource.buckets.all():58        print(bucket.name)59    first_bucket = s3_resource.Bucket(name='larry-tusk-practice')60    for obj in first_bucket.objects.all():61        print(obj.key)62    '''63    # delete bucket64    first_bucket_name = 'tute-geek'65    # s3_resource.Bucket(first_bucket_name).delete()66    # s3_resource.meta.client.delete_bucket(Bucket=first_bucket_name)67    delete_all_objects(first_bucket_name)68    '''69    copy_to_bucket('firstboto99725a2b-44e0-482b-a1bb-92a64ae06040', 'tute-geek', '883971firstfile.txt')70    # ACL control adjustment71    second_file_name = create_temp_file(400, 'secondfile.txt', 's')72    second_object = s3_resource.Object('tute-geek', second_file_name)73    second_object.upload_file(second_file_name, ExtraArgs={'ACL':'public-read'})74    second_object_acl = second_object.Acl()75    print(second_object_acl.grants)76    response = second_object_acl.put(ACL='private')77    print(second_object_acl.grants)78    # encryption and storage class adjustment79    third_file_name = create_temp_file(300, 'thirdfile.txt','t')80    third_object = s3_resource.Object('tute-geek', third_file_name)81    third_object.upload_file(third_file_name, ExtraArgs={82        'ServerSideEncryption': 'AES256',83        'StorageClass': 'STANDARD_IA'})84    print(third_object.server_side_encryption)85    print(third_object.storage_class)86    # bucket and object versioning87    first_file_name = create_temp_file(300, 'firstfile.txt', 'f')88    second_file_name = create_temp_file(400, 'secondfile.txt', 's')89    third_file_name = create_temp_file(400, 'thirdfile.txt', 's')90    s3_resource.Object('tute-geek', first_file_name).upload_file(first_file_name)91    s3_resource.Object('tute-geek', first_file_name).upload_file(second_file_name)92    s3_resource.Object('tute-geek', third_file_name).upload_file(third_file_name)93    def enable_bucket_versioning(bucket_name):94        bkt_versioning = s3_resource.BucketVersioning(bucket_name)95        bkt_versioning.enable()96        print(bkt_versioning.status)97    print(enable_bucket_versioning('tute-geek'))98    print(s3_resource.Object('tute-geek', first_file_name).version_id)99    '''100if __name__ == '__main__':...

s3_mock.py

Source:s3_mock.py

1from contextlib import contextmanager2# Use a context manager to help handle setup/teardown automatically before/after tests are run3@contextmanager4def s3_setup(s3_resource, bucketName):5    s3_resource.create_bucket(Bucket=bucketName)6    yield7def list_buckets(s3_resource):8    response = s3_resource.bucket.all()9    return response10def list_objects(s3_resource, bucketName):11    response = s3_resource.Bucket(bucketName).objects.all()12    return response13def upload_file(s3_resource, bucketName, fileName, s3Path, metadata):14    if fileName and s3Path:15        response = s3_resource.Bucket(bucketName).upload_file(fileName, s3Path, ExtraArgs={"Metadata": metadata})16    return response17def put_object(s3_resource, bucketName, objectBody, s3Path):18    if objectBody and s3Path:19        response = s3_resource.Bucket(bucketName).put_object(objectBody, s3Path)20    return response21def generate_presigned_url(s3_client, bucketName, s3Path, clientMethod='get_object', expiredIn=3600):22    response = s3_client.generate_presigned_url(23        ClientMethod=clientMethod,24        Params={'Bucket': bucketName, 'Key': s3Path},25        ExpiresIn=expiredIn26    )27    return response28def S3_Bucket_Mock(s3_resource, bucketName, initialFiles=[]):29    with s3_setup(s3_resource, bucketName):30        for initialFile in initialFiles:31            upload_file(s3_resource, bucketName, initialFile.get('filename'), initialFile.get('s3Path'), initialFile.get('metadata') or {})...

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.