Best Python code snippet using localstack_python
test_aws.py
Source:test_aws.py  
1from datetime import datetime2import boto33import pytest4from botocore.stub import ANY, Stubber5from lektorium.aws import AWS, BUCKET_POLICY_TEMPLATE6def test_create_s3_bucket():7    bucket_name = AWS.S3_PREFIX + 'foo'8    stub_response = {'ResponseMetadata': {'HTTPStatusCode': 200}}9    expected_params = {'Bucket': bucket_name}10    client = boto3.client('s3')11    stubber = Stubber(client)12    stubber.add_response('create_bucket', stub_response, expected_params)13    aws = AWS()14    aws.s3_client = client15    with stubber:16        response = aws.create_s3_bucket('foo')17    assert response == bucket_name18def test_open_bucket_access():19    bucket_name = AWS.S3_PREFIX + 'foo'20    stub_response = {'ResponseMetadata': {'HTTPStatusCode': 204}}21    expected_params_1 = {'Bucket': bucket_name}22    expected_params_2 = {23        'Bucket': bucket_name,24        'Policy': BUCKET_POLICY_TEMPLATE.format(bucket_name=bucket_name),25    }26    client = boto3.client('s3')27    stubber = Stubber(client)28    stubber.add_response(29        'delete_public_access_block',30        stub_response,31        expected_params_1,32    )33    stubber.add_response(34        'put_bucket_policy',35        stub_response,36        expected_params_2,37    )38    stubber.add_response(39        'put_bucket_website',40        {'ResponseMetadata': {'HTTPStatusCode': 200}},41        dict(42            Bucket=bucket_name,43            WebsiteConfiguration=dict(44                ErrorDocument=dict(45                    Key='404.html',46                ),47                IndexDocument=dict(48                    Suffix='index.html',49                ),50            ),51        ),52    )53    aws = AWS()54    aws.s3_client = client55    with stubber:56        aws.open_bucket_access(bucket_name)57def test_open_bucket_access_timeout():58    bucket_name = AWS.S3_PREFIX + 'foo'59    stub_response = {'ResponseMetadata': {'HTTPStatusCode': 404}}60    expected_params_1 = {'Bucket': bucket_name}61    client = boto3.client('s3')62    stubber = Stubber(client)63    stubber.add_response(64        'delete_public_access_block',65        stub_response,66        expected_params_1,67    )68    aws = AWS()69    aws.s3_client = client70    aws.SLEEP_TIMEOUT = 0.0171    with pytest.raises(Exception):72        with stubber:73            aws.open_bucket_access(bucket_name)74def test_create_cloudfront_distribution():75    bucket_name = AWS.S3_PREFIX + 'foo'76    region = boto3.client('s3').meta.region_name77    origin_domain = f'{bucket_name}.s3-website-{region}.{AWS.S3_SUFFIX}'78    distribution_id = 'bar'79    domain_name = 'buzz'80    stub_response = {81        'ResponseMetadata': {'HTTPStatusCode': 201},82        'Distribution': {83            'Id': distribution_id,84            'DomainName': domain_name,85            'ARN': '',86            'Status': '',87            'LastModifiedTime': datetime.now(),88            'InProgressInvalidationBatches': 1,89            'ActiveTrustedSigners': {'Quantity': 0, 'Enabled': False},90            'DistributionConfig': {91                'CallerReference': '',92                'Origins': {'Quantity': 1, 'Items': [{93                    'Id': '',94                    'DomainName': '',95                    'S3OriginConfig': {'OriginAccessIdentity': ''},96                }]},97                'DefaultCacheBehavior': {98                    'TargetOriginId': '',99                    'ForwardedValues': {100                        'QueryString': False,101                        'Cookies': {'Forward': 'all'},102                    },103                    'TrustedSigners': {'Quantity': 0, 'Enabled': False},104                    'ViewerProtocolPolicy': '',105                    'MinTTL': 1,106                },107                'Comment': '',108                'Enabled': True,109            },110        },111    }112    expected_params = dict(113        DistributionConfig=dict(114            CallerReference=ANY,115            Comment=ANY,116            Enabled=True,117            Origins=dict(118                Quantity=1,119                Items=[dict(120                    Id=ANY,121                    DomainName=origin_domain,122                    CustomOriginConfig=dict(123                        HTTPPort=80,124                        HTTPSPort=443,125                        OriginProtocolPolicy='http-only',126                    ),127                )],128            ),129            DefaultCacheBehavior=dict(130                TargetOriginId=ANY,131                ViewerProtocolPolicy='redirect-to-https',132                TrustedSigners=dict(Quantity=0, Enabled=False),133                ForwardedValues=dict(134                    Cookies={'Forward': 'all'},135                    Headers=dict(Quantity=0),136                    QueryString=False,137                    QueryStringCacheKeys=dict(Quantity=0),138                ),139                MinTTL=ANY,140            ),141        ),142    )143    client = boto3.client('cloudfront')144    stubber = Stubber(client)145    stubber.add_response('create_distribution', stub_response, expected_params)146    aws = AWS()147    aws.cloudfront_client = client148    with stubber:149        response = aws.create_cloudfront_distribution(bucket_name)...responses.py
Source:responses.py  
...19            return self.get_public_access_block(headers)20        elif request.method == "PUT":21            return self.put_public_access_block(request, headers)22        elif request.method == "DELETE":23            return self.delete_public_access_block(headers)24    def get_public_access_block(self, headers):25        account_id = headers["x-amz-account-id"]26        public_block_config = s3control_backend.get_public_access_block(27            account_id=account_id,28        )29        template = self.response_template(S3_PUBLIC_ACCESS_BLOCK_CONFIGURATION)30        return 200, {}, template.render(public_block_config=public_block_config)31    def put_public_access_block(self, request, headers):32        account_id = headers["x-amz-account-id"]33        pab_config = self._parse_pab_config(request.body)34        s3control_backend.put_public_access_block(35            account_id, pab_config["PublicAccessBlockConfiguration"]36        )37        return 201, {}, json.dumps({})38    def delete_public_access_block(self, headers):39        account_id = headers["x-amz-account-id"]40        s3control_backend.delete_public_access_block(account_id=account_id,)41        return 204, {}, json.dumps({})42    def _parse_pab_config(self, body):43        parsed_xml = xmltodict.parse(body)44        parsed_xml["PublicAccessBlockConfiguration"].pop("@xmlns", None)...s3_delete_public_access_policy.py
Source:s3_delete_public_access_policy.py  
...6    return account_id7def delete_bucket_policy(profile):8    session = boto3.Session(profile_name=profile)9    client = session.client("s3control", region_name="us-east-1")10    response = client.delete_public_access_block(AccountId=get_account_id(profile))11    print(json.dumps(response, indent=4))12if (13    __name__ == "__main__"14):  # takes profile_name as an argument. this could be done simpler, but I use profiles....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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
