Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use task_upload_file method in autotest
Best Python code snippet using autotest_python
forms.py
Source:forms.py 
1from django import forms2from django.contrib.auth import get_user_model3from django.core.mail import send_mail4from django.core.urlresolvers import resolve5from crispy_forms.helper import FormHelper6from crispy_forms.layout import Layout, Field, Submit, Fieldset, ButtonHolder, HTML, Button7from crispy_forms.bootstrap import FieldWithButtons, StrictButton8from django.forms.fields import ChoiceField9import random10import string11import hashlib12from .models import Task, Participation13from quiz.models import Quiz, AnswerKey, Answer14from label.models import Label15from post.models import Post16import os17import codecs18import csv19import unicodecsv20from io import TextIOWrapper21CHOICES = (('T', 'True',), ('F', 'False',))22QCHOICES = (('create', 'Create New Quiz',), ('choose', 'Choose From Available Quizes',))23#class PostModelForm(forms.ModelForm):24# class Meta:25 # model = Post26 # def __init__(self, *args, **kwargs):27 # forms.ModelForm.__init__(self, *args, **kwargs)28 #if 'initial' in kwargs:29 # self.fields['post_list'].queryset = Post.objects.filter(Post.id=initial.id)#Post.avail.all()30#############################Create Task#######################################31class CreateTaskForm(forms.Form):32 33 def __init__(self, *args, **kwargs):34 super(CreateTaskForm, self).__init__(*args, **kwargs)35 self.fields['Task Title'] = forms.CharField(max_length=50)36 self.fields['Task Description'] = forms.CharField(widget=forms.Textarea, required=False)37 self.fields['Min Number of Posts'] = forms.IntegerField(required=False)38 self.fields['Min Number of Labelers'] = forms.IntegerField(required=False)39 self.fields['Upload Task Posts and Labels'] = forms.FileField()40 41 self.fields['Choose Quiz'] =forms.ModelChoiceField(queryset=Quiz.objects.all(), required=False)##forms.CharField(max_length=5, widget=forms.Select(choices=QCHOICES))42 #Quiz.objects.filter(creator=self.request.user.pk)43 #Quiz.objects.filter(creator=self.request.user.pk)44 #Quiz.objects.all()45 self.fields['Quiz Title'] = forms.CharField(max_length=50, required=False)46 self.fields['Quiz Description'] = forms.CharField(widget=forms.Textarea, required=False)47 self.fields['Number of Questions'] = forms.IntegerField(required=False)48 self.fields['Pass Mark'] = forms.IntegerField(max_value=100, required=False)49 50 self.fields['Label Posts in Random Order'] = forms.CharField(max_length=5, widget=forms.Select(choices=CHOICES))51#widget=forms.RadioSelect(attrs={'class': 'Radio'}), choices=('True','False')52 self.fields['Upload Quiz Posts and Labels'] = forms.FileField(required=False)53 self.fields['Participating Labelers'] = forms.CharField(widget=forms.Textarea, required=False)54 self.helper = FormHelper()55 self.helper.form_id = 'create_task_form'56 self.helper.form_method = 'POST'57 self.helper.form_action = '/task/create_task/'58 self.helper.layout = Layout(59 HTML("""<div class="panel panel-info">60 <div class="panel-heading">61 Step 1: Create Task62 </div>63 <div class="panel-body">64 """),65 Field('Task Title', placeholder="Enter the task title..."),66 Field('Task Description', placeholder="Write some description about the task..."),67 Field('Min Number of Posts', placeholder="Minimum number of posts to be labeled by labelers for this task"),68 Field('Min Number of Labelers', placeholder="Minimum number of labelers per post"), 69 HTML(""""""),70 Field('Label Posts in Random Order', placeholder=""),71 Field('Upload Task Posts and Labels', placeholder = ""),72 HTML("""<style> p.solid {border : 1px; display: inline-block ; border-style: solid;} </style>73<!--table style="border: 1px solid black;"--> 74<tr>75<td>76<p class="solid">77<b>Note</b> : Upload a csv file with <a href="/media/sample_survey1.csv" download="sample_survey1.csv">this sample format</a> <br/>[All labels separated with pipelines on the first line. <br/>Each post on a separate line after that.]78</td>79</tr>80</p>81<!--/table-->82"""),83 HTML("""</div>84 </div>"""),85 HTML("""<div class="panel panel-info">86 <div class="panel-heading">87 Step 2: Choose/Create Quizzes (Optional)88 <!--span title="You can either create a new quiz or choose one from the provided list">?</span--> 89 90 <!--img src="help.png" alt="Nature" onmouseover="helpText()" onmouseout="helpBack()"-->91 </div>92 <div class="panel-body">93 <p> <b>Note</b>: This is an optional prerequisite quiz for the task. You may select quizzes from the given list or create a new one <a href="http://dblab-rack30.cs.ucr.edu:8000/quiz/create_quiz/">here</a>.</p>94 """),95 Field('Choose Quiz', placeholder=""),96 #Field('Quiz Title', placeholder="Enter the quiz title..."),97 #Field('Quiz Description', placeholder="Write some description about the quiz..."),98 #Field('Number of Questions'),99 #Field('Pass Mark'),100 #Field('Upload Quiz Posts and Labels'),101 HTML("""</div>102 </div>"""),103 HTML("""<div class="panel panel-info">104 <div class="panel-heading">105 Step 3: Send Task106 </div>107 <div class="panel-body">108 <p> <b>Note</b>: Enter the labelers to participate in the task line by line </p>109 """),110 Field('Participating Labelers', placeholder=" labeler1@example.com \n labeler2@example.com \n ..."),111 HTML("""</div>112 </div>"""),113 )114 self.helper.add_input(Submit('create_task_submit', 'Submit', css_class='btn btn-info btn-sm pull-right'))115 def create_task(self, task_title, task_description, user, prerequisite, task_num_labelers, task_num_posts, task_random_label, task_upload_file):116 117 try:118 # task_obj = Task.objects.create(119 # #id = default,120 # title = task_title,121 # description = task_description,122 # num_posts = task_num_posts,123 # num_labelers = task_num_labelers,124 # random_label = task_random_label,125 # creator = user,126 # prerequisite = prerequisite,127 # upload_file = task_upload_file128 # )129 # task_obj.save()130 # self.id = task_obj.id131 #print (task_upload_file)132 #mypath = os.path.dirname(os.path.abspath(__file__))133 #print (mypath)134# csvfile = open(str(task_upload_file), 'r')135# csvfile = open(str(task_upload_file), 'rb')136 #print("this issss", csvfile)137 # print (os.path.abspath(csvfile))138 #csvfile = open(str(task_upload_file), 'rb')# as csvfile:139 #print (csvfile)140 #print(task_upload_file)141 142# readerR = csv.reader(TextIOWrapper(task_upload_file), delimiter='|', skipinitialspace=True)143# print(TextIOWrapper(task_upload_file))144# print (readerR)145 #svfile = open(task_upload_file, 'rb')# as csvfile:146 #rint (csvfile)147 print (task_upload_file)148 readerR = csv.reader(TextIOWrapper(task_upload_file), delimiter='|', skipinitialspace=True)149 #readerR = ''150 #task_upload_file = UTF8Recoder(task_upload_file, "utf-8")151 #csvfile = open(task_upload_file, encoding='utf-8') 152 #print(csvfile)153 #readerR = csv.reader(csvfile, delimiter='|', skipinitialspace=True)154 #for row in csv.reader(codecs.iterencode(codecs.iterdecode(task_upload_file, "utf-8"), "utf-8")):155 #yield [e.decode("utf-8") for e in row]156 #print (row)157# with open(str(task_upload_file), mode='r', encoding='utf-8', newline='') as csvarchive:158# print('sjdkashkj')159# readerR = csv.reader(csvarchive, delimiter='|', skipinitialspace=True)160# print (next(readerR)) 161 #readerR = unicodecsv.reader(TextIOWrapper(task_upload_file), encoding='utf-8', delimiter='|')162# readerF = csv.reader(TextIOWrapper(task_upload_file), delimiter='|', skipinitialspace=True)163 #reader = csv.reader(open(task_upload_file, newline=''), delimiter='|')164 print ("here")165 #print(readerR)166 # posts = readerR167 # print (posts)168 # for post in posts:169 # print (post)170 #for row in readerR:171 # yield [unicode(cell, 'utf-8') for cell in row]172 labelsF = next(readerR)173 print (labelsF)174 #random.shuffle(labelsF)175 #print(labelsF)176# for rowF in labelsF:177 #print (labelsF)178 #print (task_random_label) 179 print ('here again') 180# task_obj = Task.objects.create(181# title = task_title,182# description = task_description,183# num_posts = task_num_posts,184# num_labelers = task_num_labelers,185# random_label = task_random_label,186# creator = user,187# prerequisite = prerequisite188# )189 #print(task_random_label)190 task_obj = Task.objects.create(191 #id = default,192 title = task_title,193 description = task_description,194 num_posts = task_num_posts,195 num_labelers = task_num_labelers,196 random_label = task_random_label,197 creator = user,198 prerequisite = prerequisite,199 upload_task= task_upload_file200 )201 #task_obj.full_clean()202 task_obj.save()203 self.id = task_obj.id204 205 206 posts = readerR207 #print ('here again')208 # print(posts)209 print (posts)210 list2_shuf_temp = []211 ctr = 0212 for post in posts: 213 print(post)214 if post != [] :215 list2_shuf_temp.append(post[0])216 ctr = ctr + 1217 # print(post)218 print(task_random_label)219 if task_random_label == 'T': 220 print ('here in T')221 list1_shuf = []222 list2_shuf = []223 index_shuf = range(ctr) #len(posts))224 index_shuf1 = [i for i in range(ctr)]#len(posts))]225 print(index_shuf)226 random.shuffle(index_shuf1)227 print('random')228 for i in index_shuf1: 229 #print(labelsF)230 #temp = labelsF[i] 231 #list1_shuf.append(temp) 232 # print (list2_shuf_temp)233 temp = list2_shuf_temp[i]234 list2_shuf.append(temp)235 print(list2_shuf)236 list1_shuf = labelsF237 else : 238 list2_shuf = list2_shuf_temp239 list1_shuf = labelsF240 for label in list1_shuf:241 #print (label)242 label_object = Label.objects.create(content=label) 243 task_obj.label_list.add(label_object)244 print (label)245 #random.shuffle(posts)246 #print (posts)247 for post in list2_shuf:248 post_object = Post.objects.create(249 content = post,250 author = user)251 252 print (post)253 task_obj.post_list.add(post_object)254 255 except:256 return None257 return task_obj258 def create_quiz(self, quiz_title, quiz_description, max_posts, pass_mark, quiz_upload_file, user):259 try:260 261 #reader = csv.reader(TextIOWrapper(quiz_upload_file), delimiter='|', skipinitialspace=True)262 csvfile = open(str(quiz_upload_file), 'rb')# as csvfile:263 print (csvfile)264 readerR = csv.reader(csvfile, delimiter='|', skipinitialspace=True)265 quiz_object = Quiz.objects.create(266 title = quiz_title,267 description = quiz_description,268 max_posts = max_posts,269 pass_mark = pass_mark,270 creator = user)271 answer_key_object = AnswerKey.objects.create(quiz=quiz_object)272 label_list = []273 labelsF = next(readerR)274 275 #print(quiz_title)276 for label in labelsF:277 #print (label)278 label_object = Label.objects.create(content=label)279 label_list.append(label_object)280 quiz_object.label_list.add(label_object)281 postsF = readerR282 for post in postsF:283 #print (post)284 post_object = Post.objects.create(285 content = post[0],286 author = user)287 quiz_object.post_list.add(post_object)288 answer_object = Answer.objects.create(289 answer_key = answer_key_object,290 post = post_object,291 label = [item for item in label_list if item.content == post[1]][0])292 except:293 return None294 return quiz_object295 296 def add_participant(self, task, coder):297 Participation.objects.create(298 task=task,299 labeler=coder)300 def send_email(self, task):301 participating_coders = self.cleaned_data['Participating Labelers']302 coder_list = participating_coders.split()303 print (coder_list)304 coder_l = []305 #for coder in coder_list:306 # print (coder)307 # subject = str.format("Task created for {0}", coder)308 # message = str.format("Hi {0},\n\n\tYou have been selected to complete a task. Please start here: http://0.0.0.0:8000/account/signup/\n\nBest,\nUCIPT Team", coder)309 # print (task, coder)310 # self.add_participant(task, coder)311 # coder_l.append(coder)312 # send_mail(subject, message, 'sara.alaee@gmail.com', coder_l, fail_silently=False)313 # coder_l = []314 for coderP in coder_list:315 print (coderP)316 #subject = str.format("Task created for {0}", coderP)317 #message = str.format("Hi {0},\n\n\tYou have been selected to complete a task. Please start here: [WEBSITE_URL]\n\nBest,\nUCIPT Team", coderP)318 print (coderP)319 self.add_participant(task, coderP)320 coder_l.append(coderP)321 #send_mail(subject, message, 'sara.alaee@gmail.com', coder_l, fail_silently=False)322 coder_l = []323 #UserModel = get_user_model()324 passwrd = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(10))325 passw = hashlib.sha224(passwrd.encode('utf-8')).hexdigest()326 print(passwrd)327 print(passw)328 User = get_user_model()329 if not User.objects.filter(email=coderP) :330 print ('there') 331 User.objects.create(email=coderP)332 user = User._default_manager.get(email=coderP)333 #print (user)334 user.set_password(passwrd)335 user.save()336 subject = str.format("Task created for {0}", coderP)337 message = str.format("Hi {0},\n\n\tYou have been selected to complete a task. Please start here: http://dblab-rack30.cs.ucr.edu:8000/account/login\n Your password :{1} \nBest,\nSocial Post Analyzer System", coderP, passwrd)338 print (coderP)339 print (subject)340 #self.add_participant(task, coder)341 coder_l.append(coderP)342 #print (coder_l)343 #print ('right here')344 send_mail(subject, message, 'sara.alaee@gmail.com', coder_l, fail_silently=False)345 else : 346 print ('there else') 347 user = User._default_manager.get(email=coderP)348 print (user)349 subject = str.format("Task created for {0}", coderP)350 message = str.format("Hi {0},\n\n\tYou have been selected to complete a task. Please start here: http://dblab-rack30.cs.ucr.edu:8000/account/login\n\nBest,\nSocial Post Analyzer System", coderP)351 #print (coderP)352 #print (subject)353 #self.add_participant(task, coder)354 coder_l.append(coderP)355 print (coder_l)356 print ('right here')357 send_mail(subject, message, 'sara.alaee@gmail.com', coder_l, fail_silently=False)358 #user.set_password(passwrd)359 #user.save()360 #form = self.form_class(request.POST)361 #assert uidb64 is not None and token is not None # checked by URLconf362 #try:363 # uid = urlsafe_base64_decode(uidb64) 364 # user = UserModel._default_manager.get(pk=uid)365 366 #except (TypeError, ValueError, OverflowError, UserModel.DoesNotExist):367 # user = None368 #user.set_password(passwrd)369 #user.save()370# subject = str.format("Task created for {0}", coderP)371# message = str.format("Hi {0},\n\n\tYou have been selected to complete a task. Please start here: http://dblab-rack30.cs.ucr.edu:8000/account/login\n Your password :{1} \nBest,\nUCIPT Team", coderP, passwrd)372# print (coderP)373# print (subject)374 #self.add_participant(task, coder)375# coder_l.append(coderP)376# print (coder_l)377# print ('right here')378# send_mail(subject, message, 'sara.alaee@gmail.com', coder_l, fail_silently=False) 379################Edit Task####################################380class EditTaskForm(forms.Form):381 model = Task382 def __init__(self, *args, **kwargs):383 super(EditTaskForm, self).__init__(*args, **kwargs)384 self.fields['Task Title'] = forms.CharField(max_length=50)385 self.fields['Task Description'] = forms.CharField(widget=forms.Textarea, required=False)386 self.fields['Min Number of Posts'] = forms.IntegerField(required=False)387 self.fields['Min Number of Labelers'] = forms.IntegerField(required=False)388 self.fields['Upload Task Posts and Labels'] = forms.FileField()389 390 self.fields['Choose Quiz'] =forms.ModelChoiceField(queryset=Quiz.objects.all(), required=False)##forms.CharField(max_length=5, widget=forms.Select(choices=QCHOICES))391 self.fields['Quiz Title'] = forms.CharField(max_length=50, required=False)392 self.fields['Quiz Description'] = forms.CharField(widget=forms.Textarea, required=False)393 self.fields['Number of Questions'] = forms.IntegerField(required=False)394 self.fields['Pass Mark'] = forms.IntegerField(max_value=100, required=False)395 396 self.fields['Label Posts in Random Order'] = forms.CharField(max_length=5, widget=forms.Select(choices=CHOICES))397#widget=forms.RadioSelect(attrs={'class': 'Radio'}), choices=('True','False')398 self.fields['Upload Quiz Posts and Labels'] = forms.FileField(required=False)399 self.fields['Participating Labelers'] = forms.CharField(widget=forms.Textarea, required=False)400 title = Task.objects.filter(title=self.Task.title)401 print (title)402 self.helper = FormHelper()403 self.helper.form_id = 'edit_task_form'404 self.helper.form_method = 'POST'405 self.helper.form_action = '/task/edit_task/'406 self.helper.layout = Layout(407 HTML("""<div class="panel panel-info">408 <div class="panel-heading">409 Step 1: Create Task410 </div>411 <div class="panel-body">412 """),413 Field('Task Title', placeholder=""),414 Field('Task Description', placeholder="Write some description about the task..."),415 Field('Min Number of Posts', placeholder=""),416 Field('Min Number of Labelers', placeholder=""),417 Field('Label Posts in Random Order', placeholder=""),418 Field('Upload Task Posts and Labels', placeholder ="Upload us one post per line"),419 HTML("""</div>420 </div>"""),421 HTML("""<div class="panel panel-info">422 <div class="panel-heading">423 Step 2: Choose/Create Quiz (Optional)424 <script type="text/javascript">425 function helpText() {426 document.getElementById("help").innerHTML = " This is a help text";427 }428 function helpBack() {429 document.getElementById("help").innerHTML = "?";430 }431 </script>432 <div id="help" onmouseover="helpText()" onmouseout="helpBack()">?</div>433 </div>434 <div class="panel-body">435 <p> <b>Note</b>: This is an optional prerequisite quiz for the task. You may select quizzes from the given list or create a new one <a href="http://dblab-rack30.cs.ucr.edu:8000/quiz/create_quiz/">here</a>.</p>436 """),437 Field('Choose Quiz', placeholder=""),438 #Field('Quiz Title', placeholder="Enter the quiz title..."),439 #Field('Quiz Description', placeholder="Write some description about the quiz..."),440 #Field('Number of Questions'),441 #Field('Pass Mark'),442 #Field('Upload Quiz Posts and Labels'),443 HTML("""</div>444 </div>"""),445 HTML("""<div class="panel panel-info">446 <div class="panel-heading">447 Step 3: Send Task448 </div>449 <div class="panel-body">450 <p> <b>Note</b>: Enter the coders to participate in the task line by line </p>451 """),452 Field('Participating Labelers', placeholder=" coder1@example.com \n coder2@example.com \n ..."),453 HTML("""</div>454 </div>"""),455 )456 self.helper.add_input(Submit('edit_task_submit', 'Submit', css_class='btn btn-info btn-sm pull-right'))457 def edit_task(self, task_title, task_description, user, prerequisite, task_num_labelers, task_num_posts, task_random_label, task_upload_file):458 459 try:460 csvfile = open(str(task_upload_file), 'rb')# as csvfile:461 #print (csvfile)462 readerR = csv.reader(csvfile, delimiter='|', skipinitialspace=True)463 #print (readerR)464# readerF = csv.reader(TextIOWrapper(task_upload_file), delimiter='|', skipinitialspace=True)465 #reader = csv.reader(open(task_upload_file, newline=''), delimiter='|')466 #print (readerF)467 labelsF = next(readerR)468# for rowF in labelsF:469 #print (labelsF)470 #print (task_random_label)471 task_obj = Task.objects.create(472 title = task_title,473 description = task_description,474 num_labelers = task_num_labelers,475 num_posts = task_num_posts,476 random_label = task_random_label,477 prerequisite = prerequisite,478 creator = user)479 #print(task_random_label)480 481 for label in labelsF:482 #print (label)483 label_object = Label.objects.create(content=label) 484 task_obj.label_list.add(label_object)485 #print (label_object)486 487 posts = readerR488 for post in posts:489 post_object = Post.objects.create(490 content = post[0],491 author = user)492 #print (post)493 task_obj.post_list.add(post_object)494 except:495 return None496 return task_obj497############Update Labeler####################################498class UpdateLabelerForm(forms.Form):499 selectTask = forms.CharField(required=False)500 501 def __init__(self, *args, **kwargs):502 initial_arguments = kwargs.get('initial', None)503 updated_initial = {}504 lse = ''505 if initial_arguments:506 # We have initial arguments, fetch 'user' placeholder variable if any507 taskname = initial_arguments.get('Select Task', None)508 labelers = initial_arguments.get('Participating Labelers', None)509 #stp =labelers.decode('utf-8')510 #print (stp)511# Now update the form's initial values if user512 updated_initial['Select Task'] = taskname513 #pp = labelers.encode("utf-8")514 #pp.decode("utf-8")515 #print (pp) 516 tl = labelers.replace('\n', '')517 tl = tl.replace('b\'', '\'')518 tl = tl.replace("\'", '')519# tl = tl.replace(" ", "")520 tl = tl.replace(',','\n') 521 tl = tl.replace("[", "")522 tl = tl.replace("]","")523 tl = tl.replace(" ", "")524# tl = tl.replace(" ", "")525 #print(tl.str.decode('utf-8'))526 print (tl)527 #test = bytes(tl, encoding='utf-8') 528 #test.decode('utf-8')529 #print (test)530 #tl.decode("utf-8", "replace")531# str(tl, "utf-8", "strict")532 updated_initial['Participating Labelers'] = tl533 # You can also initialize form fields with hardcoded values534 # or perform complex DB logic here to then perform initialization535# updated_initial['comment'] = 'Please provide a comment'536 # Finally update the kwargs initial reference537 kwargs.update(initial=updated_initial)538 super(UpdateLabelerForm, self).__init__(*args, **kwargs)539 #self.fields['Task Title'] = forms.CharField(max_length=50)540 self.val = "";541 #pk = args[0]542# listar = list(args)543 #field1= '0' 544 field = 0545 k = 1546 #current_url = resolve(request.path_info).url_name547 #print(current_url)548 #for c, item in enumerate(list(args), 1):549 # # print(c, item)550# item1 = item551 # if item != None:552 # field = item553 # k = c554 # break555 556 #item1 = i.next()557 558 # i.next()559 560 #for i in list(args):561 #print(i)562 # field1= i 563 #print(k, field)564 #a = np.array(args)565 #b = [0,1]566 #print(list(a[b]))567 #num1 = None568 #field =list(args)569 #print(field.index("task new"))570 #task_name = ''571 #for num in args:572 # if(num!=None):573 #task_name = num574 #print(num)575 #print (field.pop(0))576 #print(Task.objects.get(title=field1))577 #print(Task.objects.filter(title=str(field)))578 # if i!=None:579 #print(task_name)580 #field = self.getArgs(args)581 #print (self.getArgs(args))582 #print(self.get_first(args))583 584 585 #print(self.get_first(list(args))) 586 l = list(self.get_first(args))587 #print(l , "h: ", l)588 #print([part.encode("utf8") for part in self.get_first(args)])589 #if ([part.encode("utf8") for part in self.get_first(args)] != []):590 #print("val: ")591 # self.val = [part.encode("utf8") for part in self.get_first(args)]592 #print( kwargs.get('initial', None))593 self.fields['Select Task'] = forms.CharField(required=False)594 self.fields['Select Task'].widget.attrs['readonly'] = True595#ModelChoiceField(queryset=Task.objects.filter(title=[part.encode("utf8") for part in self.get_first(args)]))596# else : 597 #print("val: ")598 #print(self.val)599# self.fields['Select Task'] = None#str(self.val)#forms.ModelChoiceField(queryset=Task.objects.filter(title=self.val)) 600 #print(flag)601 #if (task_name==''):602 #print(flag) 603 #self.fields['Select Task'] = forms.ModelChoiceField(queryset=Task.objects.filter(title=task_name))604# self.fields['Participating Coders'] =forms.ModelChoiceField(queryset=Account.objects.all())##forms.CharField(max_length=5,605 self.fields['Participating Labelers'] = forms.CharField(widget=forms.Textarea, required=False)606 607 #print (self.fields['Participating Coders'])608 self.helper = FormHelper()609 self.helper.form_id = 'update_labeler_form'610 self.helper.form_method = 'POST'611 self.helper.form_action = '/task/update_labeler/'612 self.helper.layout = Layout(613 HTML("""<div class="panel panel-info">614 <div class="panel-heading">615 Add or Remove Labeler616 </div>617 618 """),619 620 621 622 HTML(""" 623 <div class="panel-body">624 <p> <b>Note</b>: Select a task and add a labeler to it.</p>625 """),626 Field('Select Task', placeholder=""),627 Field('Participating Labelers', placeholder="coder1@example.com \n coder2@example.com \n ..."),628 629 )630 #print(self.fields['Select Task']) 631 self.helper.add_input(Submit('update_labeler_submit', 'Submit', css_class='btn btn-info btn-sm pull-right'))632 def update_labeler(self, task, participating_coders):633 634 try:635 #csvfile = open(str(task_upload_file), 'rb')# as csvfile:636 #print (csvfile)637 #readerR = csv.reader(csvfile, delimiter='|', skipinitialspace=True)638 #print (readerR)639# readerF = csv.reader(TextIOWrapper(task_upload_file), delimiter='|', skipinitialspace=True)640 #reader = csv.reader(open(task_upload_file, newline=''), delimiter='|')641 #print (readerF)642 #labelsF = next(readerR)643# for rowF in labelsF:644 #print (labelsF)645 #print (task_random_label)646 #print(task)647 #print(coder)648 #participating_coders = self.cleaned_data['Participating Coders']649 #participating_coders= participating_coders.replace("\n", "")650 coder_list = participating_coders.split()651 #print(coder_list)652 participation_obj = None653 print(task)654 for coderP in coder_list:655 print('coder' , coderP[1:])656 coderP = coderP[0:]657 #subject = str.format("Task created for {0}", coder)658 #message = str.format("Hi {0},\n\n\tYou have been selected to complete a task. Please start here: [WEBSITE_URL]\n\nBest,\nUCIPT Team", coder)659 #self.add_participant(task, coder)660 print(coderP)661 print(Task.objects.filter(title=task).values_list('id', flat=True).first())662 #print(task.pk)663 coder_l = []664 if not Participation.objects.filter(labeler=coderP , task_id=Task.objects.filter(title=task).values_list('id', flat=True).first()) : 665 print("here")666 print(coderP)667 participation_obj = Participation.objects.create(668 task_id=Task.objects.filter(title=task).values_list('id', flat=True).first(),669 labeler=coderP)670 print(participation_obj)671 passwrd = ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(10))672 passw = hashlib.sha224(passwrd.encode('utf-8')).hexdigest()673 print(passwrd.encode('utf-8'))674 print(passw)675 User = get_user_model()676 if not User.objects.filter(email=coderP) :677 print ('there') 678 User.objects.create(email=coderP)679 user = User._default_manager.get(email=coderP)680 #print (user)681 user.set_password(passwrd)682 user.save()683 subject = str.format("Task created for {0}", coderP)684 message = str.format("Hi {0},\n\n\tYou have been selected to complete a task. Please start here: http://dblab-rack30.cs.ucr.edu:8000/account/login\n Your password :{1} \nBest,\nSocial Post Analyzer System", coderP, passwrd)685 # print (coderP)686 # print (subject)687 #self.add_participant(task, coder)688 coder_l.append(coderP)689 #print (coder_l)690 #print ('right here')691 send_mail(subject, message, 'sara.alaee@gmail.com', coder_l, fail_silently=False)692 else : 693 #print ('there else') 694 user = User._default_manager.get(email=coderP)695 #print (user)696 subject = str.format("Task created for {0}", coderP)697 message = str.format("Hi {0},\n\n\tYou have been selected to complete a task. Please start here: http://dblab-rack30.cs.ucr.edu:8000/account/login\n \nBest,\nSocial Post Analyzer System", coderP)698 #print (coderP)699 #print (subject)700 #self.add_participant(task, coder)701 coder_l.append(coderP)702 #print (coder_l)703 #print ('right here')704 send_mail(subject, message, 'sara.alaee@gmail.com', coder_l, fail_silently=False)705 # user.set_password(passwrd)706 # user.save()707 #form = self.form_class(request.POST)708 #assert uidb64 is not None and token is not None # checked by URLconf709 #try:710 # uid = urlsafe_base64_decode(uidb64) 711 # user = UserModel._default_manager.get(pk=uid)712 713 #except (TypeError, ValueError, OverflowError, UserModel.DoesNotExist):714 # user = None715 #user.set_password(passwrd)716 #user.save()717# subject = str.format("Task created for {0}", coderP)718# message = str.format("Hi {0},\n\n\tYou have been selected to complete a task. Please start here: http://dblab-rack30.cs.ucr.edu:8000/account/login\n Your password :{1} \nBest,\nUCIPT Team", coderP, passwrd)719# print (coderP)720# print (subject)721 #self.add_participant(task, coder)722# coder_l.append(coderP)723# print (coder_l)724# print ('right here')725# send_mail(subject, message, 'sara.alaee@gmail.com', coder_l, fail_silently=False)726# subject = str.format("Task created for {0}", coderP)727# message = str.format("Hi {0},\n\n\tYou have been selected to complete a task. Please start here:[http://0.0.0.0:8000/account/signup]\n\nBest,\nUCIPT Team", coderP)728# coder_l.append(coderP)729# print (coder_l)730# send_mail(subject, message, 'sara.alaee@gmail.com', coder_l, fail_silently=False)731 except:732 return None733 return participation_obj734 def get_first(iterable, default=None):735 if iterable:736 for item in iterable:737 print(item)738 return item739 print(iterable)...
beah_dummy.py
Source:beah_dummy.py
...37 print '%s T:%s task_start' % (self.machine_name, task_id)38 self.rpc2.task_start(task_id)39 for i in xrange(self.task_log_chunk_count):40 print '%s T:%s task_upload_file' % (self.machine_name, task_id)41 self.rpc2.task_upload_file(task_id, 'load_test',42 'name_%s' % task_id, len(self.task_log_chunk), self.task_log_chunk_md5,43 offset_task_log, self.task_log_chunk_enc)44 offset_task_log += len(self.task_log_chunk)45 #final chunk46 print '%s T:%s task_upload_file' % (self.machine_name, task_id)47 self.rpc2.task_upload_file(task_id, 'load_test', 'name_%s' % task_id, 48 len(self.task_log_chunk), self.task_log_chunk_md5, -1, self.task_log_chunk_enc)49 #Simulate some run time50 time.sleep(self.task_duration)51 #send result52 print '%s T:%s task_result' % (self.machine_name, task_id)53 result_id = self.rpc2.task_result(task_id, 'pass_', task_name, '1079',54 '(Pass)')55 #no chunking56 print '%s T:%s TR:%s result_upload_file' % (self.machine_name, task_id, result_id)57 self.rpc2.result_upload_file(result_id, '/',58 'load_test--result', len(self.recipe_log), self.recipe_log_md5, 0, self.recipe_log_enc)59 print '%s T:%s task_stop' % (self.machine_name, task_id)60 self.rpc2.task_stop(task_id, 'stop', 'OK')61class BeahDummyManager(object):...
clouds_with_adapter.py
Source:clouds_with_adapter.py
1from datetime import datetime, timedelta2from airflow import DAG3from airflow.operators.python_operator import PythonOperator4from modules import adapter_version5#--------------------------------------------------------------------------------------6#Definition of DAG7#--------------------------------------------------------------------------------------8default_args = {9 "owner": "airflow",10 "start_date": datetime(2019, 1, 1)11}12dag = DAG("clouds_with_adapter", default_args=default_args, schedule_interval=timedelta(days=1))13#--------------------------------------------------------------------------------------14#Tasks15#--------------------------------------------------------------------------------------16task_create_file = PythonOperator(17 task_id = "create_file",18 provide_context = True,19 python_callable = adapter_version.Client.create_file,20 dag = dag21)22task_upload_file = PythonOperator(23 task_id = "upload_file",24 python_callable = adapter_version.Client.upload_file,25 dag = dag26)27task_download_file = PythonOperator(28 task_id = "download_file",29 python_callable = adapter_version.Client.download_file,30 dag = dag31)32task_wrap_data = PythonOperator(33 task_id = "wrap_data",34 provide_context = True,35 python_callable = adapter_version.Client.wrap_data,36 dag = dag37)38task_rename_file = PythonOperator(39 task_id = "rename_file",40 provide_context = True,41 python_callable = adapter_version.Client.rename_file,42 dag = dag43)44task_move_file = PythonOperator(45 task_id = "move_file",46 provide_context = True,47 python_callable = adapter_version.Client.move_file,48 dag = dag49)50#--------------------------------------------------------------------------------------51#Sequence of execution of tasks52#--------------------------------------------------------------------------------------53task_create_file.set_upstream(task_upload_file)54task_download_file.set_upstream(task_upload_file)55task_wrap_data.set_upstream(task_download_file)56task_rename_file.set_upstream(task_download_file)...
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 autotest 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?
