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!!