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

How to use switch_region method in localstack

Best Python code snippet using localstack_python

test_generic_proxy.py

Source:test_generic_proxy.py

...149@pytest.mark.parametrize("encoding", [byte_encoding, string_encoding])150def test_arn_partition_rewriting_in_response_without_region_and_without_default_region(151    encoding, switch_region152):153    with switch_region(None):154        listener = ArnPartitionRewriteListener()155        response = Response()156        response._content = encoding(157            json.dumps({"some-data-with-arn": "arn:aws-us-gov:iam::123456789012:ArnInData"})158        )159        response._status_code = 200160        response.headers = {"some-header-with-arn": "arn:aws-us-gov:iam::123456789012:ArnInHeader"}161        result = listener.return_response(162            method="POST", path="/", data="ignored", headers={}, response=response163        )164        assert result.status_code == response.status_code165        assert result.headers == {"some-header-with-arn": "arn:aws:iam::123456789012:ArnInHeader"}166        assert result.content == encoding(167            json.dumps({"some-data-with-arn": "arn:aws:iam::123456789012:ArnInData"})168        )169@pytest.mark.parametrize("encoding", [byte_encoding, string_encoding])170def test_arn_partition_rewriting_in_response_without_region_and_with_default_region(171    encoding, switch_region172):173    with switch_region("us-gov-east-1"):174        listener = ArnPartitionRewriteListener()175        response = Response()176        response._content = encoding(177            json.dumps({"some-data-with-arn": "arn:aws:iam::123456789012:ArnInData"})178        )179        response._status_code = 200180        response.headers = {"some-header-with-arn": "arn:aws:iam::123456789012:ArnInHeader"}181        result = listener.return_response(182            method="POST", path="/", data="ignored", headers={}, response=response183        )184        assert result.status_code == response.status_code185        assert result.headers == {186            "some-header-with-arn": "arn:aws-us-gov:iam::123456789012:ArnInHeader"187        }...

Terracheck.py

Source:Terracheck.py

1import boto32from python_terraform import *3import argparse4from botocore.exceptions import ParamValidationError, ClientError5def dynamo_db_backend(region, table, ak, sk):6    dynamodb_client = boto3.client('dynamodb', aws_access_key_id=ak,7                                   aws_secret_access_key=sk, region_name=region)8    # response_dynamo = dynamodb_client.describe_table(TableName="vladitest")9    try:10        create = dynamodb_client.create_table(11            AttributeDefinitions=[12                {13                    'AttributeName': 'LockID',14                    'AttributeType': 'S'15                }16            ],17            KeySchema=[18                {19                    'AttributeName': 'LockID',20                    'KeyType': 'HASH',21                }22            ],23            ProvisionedThroughput={24                'ReadCapacityUnits': 5,25                'WriteCapacityUnits': 5,26            },27            TableName=table,28        )29    except ClientError as e:30        print(e)31def artifact_bucket_creation(region, bucketname):32    s3 = boto3.resource('s3')33    if s3.Bucket(bucketname).creation_date is None:34        print("bucket %s does not exist" % bucketname)35        s32 = boto3.client('s3', region_name=region)36        location = {'LocationConstraint': region}37        s32.create_bucket(Bucket=bucketname, CreateBucketConfiguration=location)38    else:39        print("bucket %s exists" % bucketname)40def s3_backend(region, bucket1, ak, sk):41    client = boto3.client(service_name="s3", region_name=region,42                          aws_access_key_id=ak,43                          aws_secret_access_key=sk)44    location = {'LocationConstraint': region}45    try:46        create_bucket_response = client.create_bucket(Bucket=bucket1, CreateBucketConfiguration=location)47    except ClientError as e:48        print(e)49def dynamichange(filename, replacesrc, replacedst):50    try:51        with open(filename) as f:52            s = f.read()53            if replacesrc not in s:54                print('"{replacesrc}" not found in {filename}.'.format(**locals()))55                return56        # Safely write the changed content, if found in the file57        with open(filename, 'w') as f:58            print('Changing "{replacesrc}" to "{replacedst}" in {filename}'.format(**locals()))59            s = s.replace(replacesrc, replacedst)60            f.write(s)61    except Exception as err:62        print('[inplace_change] Error : "{err}'.format(**locals()))63        exit()64def get_aws_account_id():65    return boto3.client('sts').get_caller_identity().get('Account')66def main():67    parser = argparse.ArgumentParser()68    parser.add_argument('-r', help='region to deploy stack example eu-central-1', required=True)69    parser.add_argument('-ak', help='Access Key - Admin rights for this please ', required=True)70    parser.add_argument('-sk', help='Secret Key - Admin rights for this please', required=True)71    args = parser.parse_args()72    dynamo_db_backend(args.r, "terraform-state-db", args.ak, args.sk)73    s3_backend(args.r, "terrform-state-bucket-test", args.ak, args.sk)74    dynamichange('./envparams.tfvars',75                 "SWITCH_REGION",76                 args.r)77    dynamichange('./envparams.tfvars',78                 "SWITCH_AK",79                 args.ak)80    dynamichange('./envparams.tfvars',81                 "SWITCH_SK",82                 args.sk)83    dynamichange('./variables.tf',84                 "SWITCH_BUCKET",85                 "terrform-state-bucket-test")86    dynamichange('./variables.tf',87                 "SWITCH_KEY",88                 "tfproject-key" + args.r)89    dynamichange('./variables.tf',90                 "SWITCH_REGION",91                 args.r)92    dynamichange('./variables.tf',93                 "SWITCH_DYNAMO",94                 "terraform-state-db")95    try:96        tf = Terraform(97            working_dir='./')98        tf.init()99        tf.apply(var_file='envparams.tfvars', skip_plan=True, no_color=IsFlagged, refresh=True,100                 capture_output=False)101        # tf.destroy(var_file='envparams.tfvars',102        #        # skip_plan=True, no_color=IsFlagged, refresh=True,103        #        capture_output=False, input=False)104    except Exception as err:105        print(err)...

AWS.py

Source:AWS.py

...11        self.api = boto3.resource('ec2', region_name=self.options['region'],12                                  aws_access_key_id=self.options['access'],13                                  aws_secret_access_key=self.options['secret'])14        return self.api15    def switch_region(self, region):16        self.login(region=region)17    18    def get_regions(self, **kwargs):19        if not self.api:20            return None21        regions = self.api.meta.client.describe_regions()22        return regions['Regions']23    24    def get_instances_from_region(self, instanceid=None):25        if not self.api:26            return None27        qfilter = {}28        if instanceid:29            qfilter = {30                'Filters' : [ {'Name': 'instance-id', 'Values': [instanceid]} ],31            }32        try:33            output = []34            for i in self.api.instances.filter(**qfilter):35                name = ''36                for x in i.tags:37                    if x['Key'] == 'Name':38                        name = x['Value']39                        break40                instance = {41                    'name': name,42                    'instance': i.id,43                    'state': i.state['Name'],44                    'type': i.instance_type,45                    'region': i.placement['AvailabilityZone'],46                    'address_private': i.private_ip_address,47                    'address_public': i.public_ip_address,48                    'hypervisor': i.hypervisor,49                    'hypervisor_type': i.virtualization_type,50                }51                output.append(instance)52            return output53        except:54            return []55    def get_instances(self, instanceid=None):56        output = []57        current_region = self.options['region']58        for i in self.get_regions():59            self.switch_region(i['RegionName'])60            output += self.get_instances_from_region(instanceid)61        self.switch_region(current_region)...

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.