Best Python code snippet using localstack_python
aws_api.py
Source:aws_api.py  
...106        result = {107            "Instances": []108        }109        for i in range(0, num_spot_nodes + num_od_nodes):110            gid = common.short_uid()111            if i < num_spot_nodes and 'group_id_task_spot' in config:112                gid = config['group_id_task_spot']113            if i >= num_spot_nodes and 'group_id_task_od' in config:114                gid = config['group_id_task_od']115            inst = {116                "Status": {117                    "State": "RUNNING",118                },119                "Id": common.short_uid(),120                "InstanceGroupId": gid,121                "Ec2InstanceId": common.short_uid(),122                "PublicDnsName": "",123                "PrivateDnsName": "testhost-%s" % common.short_uid(),124                "PublicIpAddress": LOCALHOST,125                "PrivateIpAddress": LOCALHOST126            }127            result['Instances'].append(inst)128    elif target == 'ElasticMapReduce.DescribeCluster':129        request = json.loads(req.data)130        cluster_id = request['ClusterId']131        result = {132            "Cluster": {133                "Name": "TestClusterName",134                "Id": cluster_id,135                "MasterPublicDnsName": "testhost-%s" % common.short_uid(),136                "ReleaseLabel": "emr-4.1.0",137                "Status": {138                    "State": "RUNNING"139                },140                "Ec2InstanceAttributes": {141                    "EmrManagedMasterSecurityGroup": "sg-%s" % common.short_uid(),142                    "EmrManagedSlaveSecurityGroup": "sg-%s" % common.short_uid(),143                    "Ec2SubnetId": "subnet-%s" % common.short_uid(),144                    "IamInstanceProfile": "EMR_EC2_DefaultRole",145                    "Ec2KeyName": "testKey",146                    "Ec2AvailabilityZone": "us-east-1b"147                }148            }149        }150    elif target == 'ElasticMapReduce.ListInstanceGroups':151        result = {152            "InstanceGroups": []153        }154        for t in ['master', 'core', 'task_od', 'task_spot']:155            key = 'group_id_%s' % t156            group_id = "%s" % (config[key] if key in config else 'group_%s' % t)157            g = {...btrfs.py
Source:btrfs.py  
...114        if type(self) != type(other):115            return False116        return dict(self) == dict(other)117    @property118    def short_uid(self):119        return self.uid.split('-')[0]120    @property121    def path(self):122        return os.path.join(self.fs.mountpoint, self.name)123    @property124    def subvolumes(self):125        return self.fs.find_subvolume_by_prefix(f"{self.name}/")126    def delete(self):127        for vol in self.subvolumes:128            vol.delete()129        try:130            sh.btrfs.subvolume.delete(self.path)131        except sh.ErrorReturnCode_1 as e:132            stderr = e.stderr.decode('utf-8')...fabfile.py
Source:fabfile.py  
...13# allow running more then one, set this env variable to seperate between instances14fab_code = os.environ.get('FAB_CODE', 'dj_default')15def short_hash(h):16    return hashlib.md5(h).hexdigest()[:10]17def short_uid():18    return uuid.uuid4().hex[:6]19@task20@parallel21def host_type():22    print env.host23    run('uname -a')24@task25@parallel26def copy():27    # TODO: maybe use rsync here28    run('mkdir -p /tmp/{fab_code}/'.format(fab_code=fab_code))29    put('../target/scala-2.11/{}'.format(target), '/tmp/{fab_code}/'.format(fab_code=fab_code))30    # if program_jar is not None:31    #     put(program_jar, '/tmp/')32@parallel33def start_client():34    run("tmux new -d -s dj-session-{fab_code}-{hash} 'java -Xmx2g -Ddj.cluster_seed={hosts} -jar /tmp/{fab_code}/{target} -mode client -cluster_conn hazelcast -cluster_code {code} &>/tmp/dj-log-{fab_code}-{short_code}-{hash}' > /dev/null".format(35        target=target,36        #id=env.hosts.index(env.host),37        hosts=env.dj_hosts,38        code=code,39        hash=short_uid(),40        short_code=code[:6],41        fab_code=fab_code,42    ))43def start_master():44    run("tmux new -d -s dj-session-master-{fab_code} 'java {extra_java_args} -Xmx2g -Ddj.cluster_seed={hosts} -jar /tmp/{fab_code}/{target} -mode master -cluster_conn hazelcast -cluster_code {code} -cp /tmp/{fab_code}/{program_jar} -maincls {main_cls} -djit {djit_cls} -debug_clazz_bytecode /tmp/djcls-{fab_code}/  &>/tmp/dj-log-master-{fab_code}-{short_code}-{hash}' > /dev/null".format(45        target=target,46        code=code,47        hosts=env.dj_hosts,48        program_jar=program_jar.split('/')[-1] if program_jar is not None else target,49        main_cls=main_cls,50        djit_cls=djit_cls,51        hash=short_uid(),52        short_code=code[:6],53        fab_code=fab_code,54        extra_java_args=extra_java_args,55    ))56@task57@runs_once58def start_remote():59    env.dj_hosts = ','.join([h.split('@')[-1] for h in env.hosts])60    master_r = execute(start_master, hosts=[env.hosts[0]])61    time.sleep(7)62    client_r = execute(start_client, hosts=env.hosts[1:])63@task64@runs_once65def start():66    env.dj_hosts = ','.join([h.split('@')[-1] for h in env.hosts])67    client_r = execute(start_client, hosts=env.hosts)68    time.sleep(7)69    local("tmux new -d -s dj-session-master-{fab_code} 'java {extra_java_args} -Xmx2g -Ddj.cluster_seed={hosts} -jar ../target/scala-2.11/{target} -mode master -cluster_conn hazelcast -cluster_code {code} -cp ../target/scala-2.11/{target} -maincls {main_cls} -djit {djit_cls} -debug_clazz_bytecode /tmp/djcls/ &>>/tmp/dj-log-master-{fab_code}' > /dev/null".format(70        target=target,71        code=code,72        hosts=env.dj_hosts,73        #program_jar=program_jar.split('/')[-1] if program_jar is not None else target,74        main_cls=main_cls,75        djit_cls=djit_cls,76        hash=short_uid(),77        short_code=code[:6],78        fab_code=fab_code,79        extra_java_args=extra_java_args,80    ))81def stop_master():82    run('tmux kill-session -t dj-session-master-{fab_code}'.format(fab_code=fab_code))83@task84@runs_once85def stop():86    local('tmux kill-session -t dj-session-master-{fab_code}'.format(fab_code=fab_code))87@task88@runs_once89def stop_remote():90    execute(stop_master, hosts=[env.hosts[0]])...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!!
