KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use root_task method in locust
Best Python code snippet using locust
job_submit.py
Source:job_submit.py 
1from base import job_desc_pb22from base import task_desc_pb23from base import reference_desc_pb24from google.protobuf import json_format5import httplib, urllib, re, sys, random6import binascii7import time8import shlex9if len(sys.argv) < 4:10 print "usage: job_submit.py <coordinator hostname> <web UI port> " \11 "<task binary> [<args>] [<job name>] [<input object>]"12 sys.exit(1)13hostname = sys.argv[1]14port = int(sys.argv[2])15job_desc = job_desc_pb2.JobDescriptor()16job_desc.uuid = "" # UUID will be set automatically on submission17if len(sys.argv) > 5:18 job_desc.name = sys.argv[5]19else:20 job_desc.name = "anonymous_job_at_%d" % (int(time.time()))21job_desc.root_task.uid = 022job_desc.root_task.name = "root_task"23job_desc.root_task.state = task_desc_pb2.TaskDescriptor.CREATED24job_desc.root_task.binary = sys.argv[3]25job_desc.root_task.priority = 526job_desc.root_task.resource_request.cpu_cores = 0.127job_desc.root_task.resource_request.ram_cap = 13107228job_desc.root_task.resource_request.net_tx_bw = 029job_desc.root_task.resource_request.net_rx_bw = 030job_desc.root_task.resource_request.disk_bw = 031if len(sys.argv) > 4:32 job_desc.root_task.args.extend(shlex.split(sys.argv[4]))33#job_desc.root_task.args.append("--v=2")34job_desc.root_task.inject_task_lib = True35if len(sys.argv) == 7:36 input_id = binascii.unhexlify(sys.argv[6])37else:38 input_id = binascii.unhexlify('feedcafedeadbeeffeedcafedeadbeeffeedcafedeadbeeffeedcafedeadbeef')39output_id = binascii.unhexlify('db33daba280d8e68eea6e490723b02cedb33daba280d8e68eea6e490723b02ce')40output2_id = binascii.unhexlify('feedcafedeadbeeffeedcafedeadbeeffeedcafedeadbeeffeedcafedeadbeef')41job_desc.output_ids.append(output_id)42job_desc.output_ids.append(output2_id)43input_desc = job_desc.root_task.dependencies.add()44input_desc.id = input_id45input_desc.scope = reference_desc_pb2.ReferenceDescriptor.PUBLIC46input_desc.type = reference_desc_pb2.ReferenceDescriptor.CONCRETE47input_desc.non_deterministic = False48input_desc.location = "blob:/tmp/fib_in"49final_output_desc = job_desc.root_task.outputs.add()50final_output_desc.id = output_id51final_output_desc.scope = reference_desc_pb2.ReferenceDescriptor.PUBLIC52final_output_desc.type = reference_desc_pb2.ReferenceDescriptor.FUTURE53final_output_desc.non_deterministic = True54final_output_desc.location = "blob:/tmp/out1"55final_output2_desc = job_desc.root_task.outputs.add()56final_output2_desc.id = output2_id57final_output2_desc.scope = reference_desc_pb2.ReferenceDescriptor.PUBLIC58final_output2_desc.type = reference_desc_pb2.ReferenceDescriptor.FUTURE59final_output2_desc.non_deterministic = True60final_output2_desc.location = "blob:/tmp/out2"61jd_as_json = json_format.MessageToJson(job_desc)62params = urllib.urlencode({'jd': jd_as_json })63print "SUBMITTING job with parameters:"64print jd_as_json65print ""66try:67 headers = {"Content-type": "application/x-www-form-urlencoded"}68 conn = httplib.HTTPConnection("%s:%s" % (hostname, port))69 conn.request("POST", "/job/submit/", params, headers)70 response = conn.getresponse()71except Exception as e:72 print "ERROR connecting to coordinator: %s" % (e)73 sys.exit(1)74data = response.read()75match = re.search(r"([0-9a-f\-]+)", data, re.MULTILINE | re.S | re.I | re.U)76print "----------------------------------------------"77if match and response.status == 200:78 job_id = match.group(1)79 print "JOB SUBMITTED successfully!\nJOB ID is %s\nStatus page: " \80 "http://%s:%d/job/status/?id=%s" % (job_id, hostname, port, job_id)81else:82 print "ERROR submitting job -- response was: %s (Code %d)" % (response.reason,83 response.status)84print "----------------------------------------------"...
hdfs_submit.py
Source:hdfs_submit.py
1from base import job_desc_pb22from base import task_desc_pb23from base import reference_desc_pb24from google.protobuf import text_format5import httplib, urllib, re, sys, random6import binascii7import time8import shlex9if len(sys.argv) < 4:10 print "usage: job_submit.py <coordinator hostname> <web UI port> " \11 "<job name> <input file> <args>"12 sys.exit(1)13hostname = sys.argv[1]14port = int(sys.argv[2])15job_desc = job_desc_pb2.JobDescriptor()16job_desc.uuid = "" # UUID will be set automatically on submission17job_desc.name = sys.argv[3]18job_desc.root_task.uid = 019job_desc.root_task.name = "root_task"20job_desc.root_task.state = task_desc_pb2.TaskDescriptor.CREATED21job_desc.root_task.binary = "/usr/bin/hadoop"22job_desc.root_task.priority = 523job_desc.root_task.resource_request.cpu_cores = 0.124job_desc.root_task.resource_request.ram_cap = 13107225job_desc.root_task.resource_request.net_tx_bw = 026job_desc.root_task.resource_request.net_rx_bw = 027job_desc.root_task.resource_request.disk_bw = 028if len(sys.argv) > 5:29 job_desc.root_task.args.extend(shlex.split(sys.argv[5]))30#job_desc.root_task.args.append("--v=2")31job_desc.root_task.inject_task_lib = True32input_desc = job_desc.root_task.dependencies.add()33input_desc.id = binascii.unhexlify('feedcafedeadbeeffeedcafedeadbeeffeedcafedeadbeeffeedcafedeadbeef')34input_desc.location = sys.argv[4]35output_id = binascii.unhexlify('db33daba280d8e68eea6e490723b02cedb33daba280d8e68eea6e490723b02ce')36job_desc.output_ids.append(output_id)37input_desc.scope = reference_desc_pb2.ReferenceDescriptor.PUBLIC38input_desc.type = reference_desc_pb2.ReferenceDescriptor.CONCRETE39input_desc.non_deterministic = False40final_output_desc = job_desc.root_task.outputs.add()41final_output_desc.id = output_id42final_output_desc.scope = reference_desc_pb2.ReferenceDescriptor.PUBLIC43final_output_desc.type = reference_desc_pb2.ReferenceDescriptor.FUTURE44final_output_desc.non_deterministic = True45final_output_desc.location = "blob:/tmp/out1"46#params = urllib.urlencode({'test': text_format.MessageToString(job_desc)})47params = 'jd=%s' % text_format.MessageToString(job_desc)48print "SUBMITTING job with parameters:"49print params50print ""51try:52 headers = {"Content-type": "application/x-www-form-urlencoded"}53 conn = httplib.HTTPConnection("%s:%s" % (hostname, port))54 conn.request("POST", "/job/submit/", params, headers)55 response = conn.getresponse()56except Exception as e:57 print "ERROR connecting to coordinator: %s" % (e)58 sys.exit(1)59data = response.read()60match = re.search(r"([0-9a-f\-]+)", data, re.MULTILINE | re.S | re.I | re.U)61print "----------------------------------------------"62if match and response.status == 200:63 job_id = match.group(1)64 print "JOB SUBMITTED successfully!\nJOB ID is %s\nStatus page: " \65 "http://%s:%d/job/status/?id=%s" % (job_id, hostname, port, job_id)66else:67 print "ERROR submitting job -- response was: %s (Code %d)" % (response.reason,68 response.status)69print "----------------------------------------------"...
utils.py
Source:utils.py
1# -*- coding:utf-8 -*-2# coding=<utf8>3from todoes.models import Note, Task, RegularTask, ProblemByWorker4# меÑод поÑÑоÑÐ¾ÐµÐ½Ð¸Ñ Ð´ÐµÑева замеÑок5def build_note_tree(root_note,notes,current_indent):6 childrens = Note.objects.filter(parent_note=root_note).\7 order_by('timestamp')8 for note in childrens:9 notes.append(note_with_indent(note,current_indent))10 build_note_tree(note,notes,current_indent+1)11# клаÑÑ Ð´Ð»Ñ Ð·Ð°Ð¼ÐµÑки Ñ Ð¾ÑÑÑÑпом12class note_with_indent():13 def __init__(self, note, indent):14 self.note = note.note15 self.id = note.id16 self.author = note.author17 self.timestamp = note.timestamp18 self.indent = '◌'*indent19 self.indent_pix = 4*indent20# полÑÑение вÑеÑ
замеÑок Ð´Ð»Ñ Ð·Ð°Ñвки21def get_all_notes(root_note,notes):22 childrens = Note.objects.filter(parent_note=root_note).\23 order_by('timestamp')24 for note in childrens:25 notes.append(note)26 get_all_notes(note,notes)27def build_tasks_tree(root_task,tasks,current_indent):28 # class_dict = {str(RegularTask):RegularTask,29 # str(Task):Task}30 if str(root_task.__class__)==str(Task):31 childrens = Task.objects.filter(parent_task=root_task).order_by('start_date')32 elif str(root_task.__class__)==str(RegularTask):33 childrens = Task.objects.filter(parent_regular_task=root_task).order_by('start_date')34 for task in childrens:35 task.indent_pix = 4*current_indent36 tasks.append(task)37 build_tasks_tree(task,tasks,current_indent+1)38def update_weigth():39 all_task = Task.objects.exclude(pbw__isnull=True)40 counts = dict()41 total_count = len(all_task)42 for task in all_task:43 try:44 counts[task.pbw.id]+=145 except:46 counts[task.pbw.id]=147 for k,v in counts.items():48 # counts[k]=v/float(total_count)49 pbw = ProblemByWorker.objects.get(id=k)50 pbw.weight = v/float(total_count)51 pbw.save()52 # print counts...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run locust automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
