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