How to use root_task1 method in locust

Best Python code snippet using locust

generate_heir.py

Source:generate_heir.py

1import sys2import subprocess3import re4import os5import platform6import networkx as nx7import json8import matplotlib.pyplot as plt9from graphviz import Digraph10#  1. Merging hierarchy for divide nodes by adding variable names in the nodes     2. reading structure from json11#act_code=[]12#act_string=raw_input()13#l=len(codes)14#for i in range(0,l):15#	act_code.append(int(codes[i]))16#var_string=raw_input()17#var_name=var_string.split(" ");18#var_range=[]19#var_string=raw_input()20#var_codes=[]21#cds=var_string.split(" ");22#l=len(cds)23#for i in range(0,l):24#	var_codes.append(int(cds[i]))25def make_graph(hierarchy, iteration):26	global actions27	global act_back28	iteration=iteration+".gv"29	d=Digraph('G',filename=iteration)30	for item in hierarchy:31		d.node(str(item))32	for item in hierarchy:33		if(hierarchy[item]["typ"]!="Actions"):34			for i in range(0,len(hierarchy[item]["children"])):35				d.edge(str(item),str(hierarchy[item]["children"][i]))36		else:37			for i in range(0,len(hierarchy[item]["value"])):38				d.edge(str(item),str(act_back[str(hierarchy[item]["value"][i])]))39	d.format='pdf'40	d.render()41def merge_hierarchy(hierarchy1 ,hierarchy2, root_task1,root_task2):42	if hierarchy1[root_task1]["typ"]=="Actions":43		hierarchy1[root_task1]["value"]=list(set(hierarchy1[root_task1]["value"]+hierarchy2[root_task2]["value"]))44		return 45	elif((hierarchy1[root_task1]["typ"]=="sequence" and hierarchy2[root_task1]["typ"]=="sequence") or (hierarchy1[root_task1]["typ"]=="root" and hierarchy2[root_task2]["typ"]=="root") ):46		hierarchy1[root_task1]["children"].sort()47		hierarchy2[root_task2]["children"].sort()48		diff=len(hierarchy1[root_task1]["children"])-len(hierarchy2[root_task2]["children"])49		if(diff>0):50			for i in range(0,diff):51				queue=[]52				hierarchy1[str((hierarchy2[root_task2]["children"])[i])+"_1"]=hierarchy2[(hierarchy2[root_task2]["children"])[i]]53				hierarchy1[root_task1]["children"].append(str((hierarchy2[root_task2]["children"])[i])+"_1")54				queue.append(hierarchy2[(hierarchy2[root_task2]["children"])[i]]["children"])55				while(queue):56					new_ele=queue[0]57					hierarchy1[(str(new_ele)+"_1")]=hierarchy2[new_ele]58					queue.append(hierarchy2[new_ele]["children"])59					queue.remove(new_ele)60		for i in range(0,len(hierarchy1[root_task1]["children"])):61			merge_hierarchy(hierarchy1,hierarchy2,hierarchy1[root_task1]["children"][i],hierarchy2[root_task2]["children"][i])62	else:63		hierarchy1[root_task1]["typ"]="divide"64		hierarchy2[root_task2]["typ"]="divide"65		list1=hierarchy1[root_task1]["variables"]66		list2=hierarchy2[root_task2]["variables"]67		for i in range(0,len(list1)):68			flag=069			mark_j=-170			for j in range(0,len(list2)):71				if(list1[i]==list2[j]):72					flag=173					mark_j=j74			if(flag==0):75				queue=[]76				for k in range(0,len(list1)):77					if(list1[i] in hierarchy1[hierarchy1[root_task1]["children"][k]]["variables"]):78						hierarchy2[str(hierarchy1[root_task1]["children"][k])+"_1"]=hierarchy1[hierarchy1[root_task1]["children"][k]]79						queue.append(hierarchy1[hierarchy1[root_task1]["children"][k]]["children"])80						while(queue):81							new_ele=queue[0]82							hierarchy2[(str(new_ele)+"_1")]=hierarchy1[new_ele]83							queue.append(hierarchy1[new_ele]["children"])84							queue.remove(new_ele)85			else:86				for k in range(0,len(list1)):87					jk=hierarchy1[hierarchy1[root_task1]["children"][k]]["variables"]88					if(list1[i] in hierarchy1[hierarchy1[root_task1]["children"][k]]["variables"]):89						for p in range(0,len(list2)):90							if(list2[mark_j] in hierarchy2[hierarchy2[root_task2]["children"][p]]["variables"]):91								merge_hierarchy(hierarchy1,hierarchy2,hierarchy1[root_task1]["children"][k],hierarchy2[root_task2]["children"][p])92		for i in range(0,len(list2)):93			flag=094			for j in range(0,len(list1)):95				if(list2[i]==list1[j]):96					flag=197			if(flag==0):98				queue=[]99				for k in range(0,len(list2)):100					if(list2[i]in hierarchy2[hierarchy2][root_task2["children"][k]]["variables"]):101						hierarchy1[str(hierarchy2[root_task2]["children"][k])+"_1"]=hierarchy2[hierarchy2[root_task2]["children"][k]]102						queue.append(hierarchy2[hierarchy2[root_task2]["children"][k]]["children"])103						while(queue):104							new_ele=queue[0]105							hierarchy1[(str(new_ele)+"_1")]=hierarchy2[new_ele]106							queue.append(hierarchy2[new_ele]["children"])107							queue.remove(new_ele)108def make_hierarchy(hierarchy,start_node, end_node, typ_e, parent_task, va):109	global my_edges110	global skeleton111	global task_number112	global actions113	variables=actions114	print start_node 115	print end_node116	if(start_node==end_node):117		return118	make_node={"children":[],"typ":typ_e, "variables":[]}119	make_node["variables"]=va120	this_task=task_number121	hierarchy[this_task]=make_node122	task_number=task_number+1123	nodes=[]124	top=0;125	for edges in skeleton:126		if(edges==start_node):127			nodes.append(edges)128			top=top+1129			nodes.append(skeleton[edges])130			break131	curr_node=nodes[1]132	while(curr_node!=" " and curr_node!=end_node and skeleton[curr_node]!=" "):133		nodes.append(skeleton[curr_node])134		curr_node=skeleton[curr_node]135	count=0136	end_list=[]137	for nod in nodes:138		if(nod!=" "):139			edges=my_edges[nod]140			for dests in edges:141				if(dests["dest"]==end_node):142					count=count+1143					end_list.append(nod);144		145	print count146	if(skeleton[start_node]==end_node):147		count=count-1148	if(count>2 or typ_e=="divide"):149		curr_start=start_node150		ed=my_edges[start_node]151		top=1152		for nod in nodes:153			for j in end_list:154				if(j==nod):155					vc=0156					for e in ed:157						if(e["dest"]==nod):158							var_ch=e["var_codes"]159							for i in range(0,len(var_ch)):160								if(var_ch[i]==1):161									vc=i+1162					make_hierarchy(hierarchy,skeleton[curr_start],nod,"divide",this_task,[vc])163					hierarchy[this_task]["variables"].append(vc)164					task_number=task_number+1165					curr_start=nod166	else:167		path_from={}                          # for marking previous node168		dist_from={}169		hierarchy[this_task]["typ"]="sequence" # for storing distance through that route                        170		for j in range(0,len(nodes)):         #marking all the nodes as maximum171			dist_from[nodes[j]]=len(nodes)+1172		dist_from[start_node]=0173		path_from[start_node]=start_node      # distance from start to start is 0174		for nod in nodes:175			edges=my_edges[nod]               # out edges from every node176			for edge in edges:177				desti=edge["dest"]              # end point for every edge178				if(desti not in nodes or dist_from[desti]>(dist_from[nod]+1)): #checking for minimum distance till now179					dist_from[desti]=dist_from[nod]+1    # updating the distances and the route information180					path_from[desti]=nod181		number_of_iter=dist_from[end_node]182		if(number_of_iter<len(nodes)-1):183			curr_dest=end_node184			for k in range(0,number_of_iter):185				rem=my_edges[path_from[curr_dest]]186				top=0187				vc=0188				for i in rem:189					if(i["dest"]==curr_dest):190						var_ch=my_edges[path_from[curr_dest]][top]["var_codes"]191						for p in range(0,len(var_ch)):192							if(var_ch[p]==1):193								vc=p+1194						del my_edges[path_from[curr_dest]][top]195					else:196						top=top+1197				make_hierarchy(hierarchy,path_from[curr_dest],curr_dest,"sequence" , this_task,[vc]) #needs to add this into hierarchy198				hierarchy[this_task]["variables"].append(vc)199				task_number=task_number+1200				curr_dest=path_from[curr_dest]201		else:202			hierarchy[this_task]["typ"]="sequence"203			hierarchy[this_task]["value"]=[]204			task_number=task_number+1205			ka1=end_node206			ka=ka1.split("_")207			new_node={"children":[],"typ":"Actions", "value": [variables[ka[0]]], "variables" :[] }208			hierarchy[this_task]["children"].append(task_number) # adding primitive actions to higher hierarchy like pickup to pickup function209			hierarchy[task_number]=new_node210			task_number=task_number+1211			nn={"children":[],"typ":"Actions","value":[],"variables" :[]}212			curr_node=skeleton[start_node]213			while(skeleton[curr_node]!=end_node and curr_node!=end_node):214				ka2=curr_node215				ka3=ka2.split("_")216				nn["value"].append(variables[ka3[0]])217				curr_node=skeleton[curr_node]218			nn["value"]=list(set(nn["value"]))219			hierarchy[task_number]=nn220			hierarchy[this_task]["children"].append(task_number)221	hierarchy[parent_task]["children"].append(this_task)222			223data=json.load(open('sample.json'))224hea=data["Header"]225header=hea.split("!")226vi=header[0]227v=vi.split(" ")228c=header[1].split(" ")229actions={}230var={}231act_back={}232num_act=len(v)233for i in range(0,len(v)):234	actions[v[i]]=int(c[i])235	act_back[c[i]]=v[i]236actions["End"]=len(v)237act_back[str(len(v))]="End"238print actions239print act_back240v=header[2].split(" ")241for i in range(0,len(v)):242	var[v[i]]=i+1243n=len(data["CatStructures"])244my_edges={}245ed=data["CatStructures"][0]["Actions"]246sk=data["CatStructures"][0]["Path"]247skeleton={}248for i in range(0,len(sk)-1):249	skeleton[sk[i]]=sk[i+1]250	my_edges[sk[i]]=[]251my_edges["End"]=[]252skeleton["End"]=" "253for i in range(1,len(ed)):254	node=ed[i]["Nodes"]255	nodes=node.split(" ")256	new_ed={}257	#new_ed={"code_for_dest": ,"dest": nodes[1], "var_codes" : [ed[i]]["Codes"], "initial_values":[0],"new_values":[0] }258	new_ed["code_for_dest"]=actions[ed[i]["Direction"]]259	new_ed["dest"]=nodes[1]260	kp1=ed[i]["Codes"]261	kp=kp1.split(",")262	new_ed["var_codes"]=[]263	for j in range(0,len(kp)):264		new_ed["var_codes"].append(int(kp[j]))265	#new_ed["var_codes"]=ed[i]["Codes"]266	new_ed["initial_values"]=[0]267	new_ed["new_values"]=[0]268	my_edges[nodes[0]].append(new_ed)269 # ex-{"North":1, "South":2 }270  # ex- { "Start": [{code_for_dest:3,dest: "East1", var_codes: [0,0,0,0,0,0,0,0,1], initial_values:[0], new_values:[1]},{code_for_dest:3,dest: "East1", var_codes: [1,1,0,0,0,0,0,0,0], initial_values:[0,0], new_values:[1,0]},{code_for_dest:5, dest:"Pickup1", var_codes: [0,0,0,0,0,0,1,0], initial_values:[0], new_values:[1]}, {code_for_dest:6,dest: "Dropoff1", var_codes: [0,0,1,1,0,0,0,0,0], initial_values:[3,0], new_values:[-1,-1]}]}271dest_to_end=[]272hierarchy1={}    # ex-{task_number: {children:[], type:}}273for edge in my_edges:  #finding all the nodes directly connected to end274	dests=my_edges[edge]275	for i in range(0,len(dests)):276		dicti=dests[i]277		if(dicti["dest"]=="End"):278			dest_to_end.append(edge)279task_number=1280make_node={"children":[],"typ":"root","variables":[]}281hierarchy1[0]=make_node282make_hierarchy(hierarchy1,"Start","End","sequence",0,[])283hierarchy2={}284my_edges.clear();285new_ed.clear();286dicti.clear();287for ki in range(1,n):288	hierarchy2.clear();289	ed=data["CatStructures"][ki]["Actions"]290	sk=data["CatStructures"][ki]["Path"]291	skeleton={}292	for i in range(0,len(sk)-1):293		skeleton[sk[i]]=sk[i+1]294		my_edges[sk[i]]=[]295	my_edges["End"]=[]296	skeleton["End"]=" "297	for i in range(1,len(ed)):298		node=ed[i]["Nodes"]299		nodes=node.split(" ")300		new_ed={}301		new_ed["code_for_dest"]=actions[ed[i]["Direction"]]302		new_ed["dest"]=nodes[1]303		kp1=ed[i]["Codes"]304		kp=kp1.split(",")305		new_ed["var_codes"]=[]306		for j in range(0,len(kp)):307			new_ed["var_codes"].append(int(kp[j]))308		new_ed["initial_values"]=[0]309		new_ed["new_values"]=[0]310		my_edges[nodes[0]].append(new_ed)311	dest_to_end=[]  312	for edge in my_edges:  313		dests=my_edges[edge]314		for i in range(0,len(dests)):315			dicti=dests[i]316			if(dicti["dest"]=="End"):317				dest_to_end.append(edge)318	task_number=1319	make_node={"children":[],"typ":"root","variables":[]}320	hierarchy2[0]=make_node321	make_hierarchy(hierarchy2,"Start","End","sequence",0,[])322	merge_hierarchy(hierarchy1,hierarchy2,0,0)323	make_graph(hierarchy1,str(ki))324	325g=nx.Graph()326print actions327print var328node_added=[]329for i in hierarchy1:330	print i331	print hierarchy1[i]332	if(i not in node_added):333		node_added.append(i)334		if(hierarchy1[i]["typ"]=="Actions"):335			g.add_node(str(i))336		else:337			g.add_node(str(i),variables=hierarchy1[i]["variables"])338	chi=hierarchy1[i]["children"]339	for j in range(0,len(chi)):340		if(j not in node_added ):341			node_added.append(j)342			if(hierarchy1[j]["typ"]!="Actions"):343				g.add_node(str(j), variables=hierarchy1[j]["variables"])344			else:345				g.add_node(str(j))346		g.add_edge(str(i),str(j))347	if hierarchy1[i]["typ"]=="Actions":348		for j in range(0,len(hierarchy1[i]["value"])):349			g.add_edge(str(i),act_back[str(hierarchy1[i]["value"][j])])...

test_taskratio.py

Source:test_taskratio.py

...4class TestTaskRatio(unittest.TestCase):5    def test_task_ratio_command(self):6        class Tasks(TaskSet):7            @task8            def root_task1(self):9                pass10            @task11            class SubTasks(TaskSet):12                @task13                def task1(self):14                    pass15                @task16                def task2(self):17                    pass18        class MyUser(User):19            tasks = [Tasks]20        ratio_dict = get_task_ratio_dict(Tasks.tasks, total=True)21        self.assertEqual(22            {...

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.