KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use iExecute method in fMBT

Best Python code snippet using fMBT_python

views.py

Source:views.py

1from django.http import Http4042from django.shortcuts import render, redirect3from django.http import HttpResponse4from django.views.generic import View5#from .forms import *6from database.models import *7from django.http import JsonResponse8import json9from .controller import *10# Create your views here.11class GoRentLoginOwnerPage(View):12	username = None13	password = None14	def get(self, request):15		return render(request, 'registers/loginOwner.html') # THIS REFERES TO TEMPLATE PATH, FIND url.py  IF YOU WANT TO LIVE ACCESS THE PAGE16	def post(self, request):17		self.username = request.POST.get("username")18		self.password = request.POST.get("password")19		ikawBahalaSaimoVariable = GoRentLogin()20		isCredentialsValid, user = ikawBahalaSaimoVariable.credentialsValidation(user_type = "RentOwner",username = self.username, password = self.password)21		print(user)22		if isCredentialsValid:23		24			# USING REDIRECT DOES NOT HAVE CONTEXT ARGUMENT THUS WE NEED TO USE SESSIONS25			request.session["user_type"] = "RentOwner"26			request.session["user_email"] = user.email27			request.session["user_password"] = user.password28			return redirect('gildo:main_view') # CALL THIS VIEW (ANG GET FUNCTION IEXECUTE ANI)29		else:30			# return HttpResponse("Fail \n inputs: " +username +", " +password +" Para rani debug dapat jud dili maka hibaw ang mo login unsa iya sayup")31			 return render(request, 'registers/loginOwner.html', {'popUpFlag':True})32			# return JsonResponse(errorMessage)33class GoRentLoginShareePage(View):34	username = None35	password = None36	def get(self, request):37		return render(request, 'registers/loginSharee.html') # THIS REFERES TO TEMPLATE PATH, FIND url.py  IF YOU WANT TO LIVE ACCESS THE PAGE 38	def post(self, request):39		self.username = request.POST.get("username")40		self.password = request.POST.get("password")41		user_object = Sharee.objects.filter(email = self.username) 42		print(user_object)43		44		if(user_object.count() == 1 and user_object[0].password == self.password): # THIS KIND OF VALIDATION IS APPLICABLE WHEN USING objects.filter()45			user_object = user_object[0]46			# USING REDIRECT DOES NOT HAVE CONTEXT ARGUMENT THUS WE NEED TO USE SESSIONS47			request.session["user_type"] = "Sharee"48			request.session["user_email"] = user_object.email49			request.session["user_password"] = user_object.password50			return redirect('gildo:main_view') # CALL THIS VIEW (ANG GET FUNCTION IEXECUTE ANI)51		else:52			# return HttpResponse("Fail \n inputs: " +username +", " +password +" Para rani debug dapat jud dili maka hibaw ang mo login unsa iya sayup")53			 return render(request, 'registers/loginSharee.html', {'popUpFlag':True})54			# return JsonResponse(errorLoginSharee)55class GoRentOwnerRegisterPage(View):	56	firstName = None57	lastName = None58	email = None59	password = None60	mobileNumber = None61	def get(self, request):62		return render(request, 'registers/ownerRegisterPage.html') # THIS REFERES TO TEMPLATE PATH, FIND url.py  IF YOU WANT TO LIVE ACCESS THE PAGE 63	def post (self,request):64		print(request.headers)65		print(str(request) + "asdadads")66		registerController = GoRentRegister()67		if 'application/x-www-form-urlencoded; charset=UTF-8' in request.headers["Content-Type"]:68			request = json.loads(request.body)69			if request['ajaxAction'] == 'validateEmail':70				emailData = {'error_code': registerController.emailValidator("RentOwner",request["checkemail"])}71				return JsonResponse(emailData)72			elif request['ajaxAction'] == 'validateContact':73				contactData = {'error_code': registerController.contactNumberValidator("RentOwner",request["checkContact"])}74				return JsonResponse(contactData)75			elif request['ajaxAction'] == 'validatePassword':76				passData = {'error_code': registerController.passwordValidator(request["checkPassword"])}77				return JsonResponse(passData)78		else:79			self.firstname = request.POST.get("firstName")80			self.lastname = request.POST.get("lastName")81			self.email = request.POST.get("email")82			self.password = request.POST.get("password")83			self.mobileNumber = request.POST.get("mobileNumber")84			birthdate = request.POST.get("birthdate")85			occupation = request.POST.get("occupation")86			obj = RentOwner(email = self.email, firstname = self.firstname, lastname = self.lastname, password = self.password, contactnumber = self.mobileNumber)87			obj.save()88			#Space(coorasdadasd, qweoiqwe, owner = obj)89			90			return redirect('registers:loginOwner_view')91class GoRentShareeRegisterPage(View):92	firstName = None93	lastName = None94	email = None95	password = None96	mobileNumber = None97	def get(self, request):98		return render(request, 'registers/shareeRegisterPage.html') # THIS REFERES TO TEMPLATE PATH, FIND url.py  IF YOU WANT TO LIVE ACCESS THE PAGE 99	def post (self,request):100		print(request.headers)101		print(str(request) + "asdadads")102		registerController = GoRentRegister()103		if 'application/x-www-form-urlencoded; charset=UTF-8' in request.headers["Content-Type"]:104			request = json.loads(request.body)105			if request['ajaxAction'] == 'validateEmail':106				emailData = {'error_code': registerController.emailValidator("Sharee",request["checkemail"])}107				return JsonResponse(emailData)108			elif request['ajaxAction'] == 'validateContact':109				contactData = {'error_code': registerController.contactNumberValidator("Sharee",request["checkContact"])}110				return JsonResponse(contactData)111			elif request['ajaxAction'] == 'validatePassword':112				passData = {'error_code': registerController.passwordValidator(request["checkPassword"])}113				return JsonResponse(passData)114		else:115			self.firstname = request.POST.get("firstName")116			self.lastname = request.POST.get("lastName")117			self.email = request.POST.get("email")118			self.password = request.POST.get("password")119			self.mobileNumber = request.POST.get("mobileNumber")120			shareeBirthdate = request.POST.get("birthdate")121			122			# shareeEmailData = {'is_shareeTaken': Sharee.objects.filter(email=shareeEmail).exists()}123			# shareeEmailData['error_shareeEmail'] = 'A user with this email already exists.'124			# if shareeEmailData['is_shareeTaken']:125			# 	return JsonResponse(shareeEmailData)126			# else:127			# space_obj = Space(coord, asdasdjasidj)128			# space_obj.save()129			# obj = Sharee( email = self.email, firstname = self.firstname, lastname = self.lastname, password = self.password, contactnumber = self.mobileNumber)130			# obj.save()131			obj = Sharee(email = self.email, firstname = self.firstname, lastname = self.lastname, password = self.password, contactnumber = self.mobileNumber)132			obj.save()133			return redirect('registers:loginSharee_view')134class GoRentLogoutPage(View):135	def get(self, request):136		request.session.clear()137		request.session.flush()138		return redirect('gildo:main_view') # CALL THIS VIEW (ANG GET FUNCTION IEXECUTE ANI)139		140class GoRentAboutPage(View):141	def get(self, request):142		return render(request, 'registers/landingPage.html')...

parallel_gaudi.py

Source:parallel_gaudi.py

1#!/usr/bin/env python2# -*- coding: utf-8 -*-3# =============================================================================4# @file ostap/parallel/parallel_gaudi.py5# This is a modified verison of the6# original <code>GaudiMP.Parallel</code> module coded by Pere MATO7# @author Pere Mato (pere.mato@cern.ch)8# 9# =============================================================================10"""11This is a modified verison of the `GaudiMP.Parallel` module by Pere MATO12GaudiMP.Parallel module:13- This module provides 'parallel' processing support for GaudiPyhton.14It is adding some sugar on top of public domain packages such as15the 'multiprocessing' or the 'pp' packages. The interface can be made16independent of the underlying implementation package.17Two main class are defined: Task and WorkManager18"""19# =============================================================================20__all__ = (21    'Task'        , ## Base class for Task 22    'WorkManager' , ## Task-manager 23    )24# =============================================================================25import sys, os, time26from   itertools                 import repeat , count27from   ostap.utils.progress_bar  import progress_bar28from   ostap.parallel.task       import Task, TaskManager 29import multiprocessing           as     MP30# =============================================================================31from    ostap.logger.logger       import getLogger32# =============================================================================33if ( 3 , 3 ) <= sys.version_info  : from collections.abc import Sized34else                              : from collections     import Sized 35# =============================================================================36logger  = getLogger('ostap.parallel.parallel_gaudi')37# =============================================================================38class pool_context :39    def __init__  ( self , pool ) :40        self.__pool = pool41    def __enter__ ( self ) :42        sys.stdout .flush ()43        sys.stderr .flush ()44        return self.__pool45    def __exit__  ( self, *_ ) :46        self.__pool.close ()47        self.__pool.join  ()48        sys.stdout .flush ()49        sys.stderr .flush ()50 51# =============================================================================52class WorkManager(TaskManager) :53    """ Class to in charge of managing the tasks and distributing them to54        the workers. They can be local (using other cores) or remote55        using other nodes in the local cluster """56    def __init__( self                     , 57                  ncpus     = 'autodetect' ,58                  ppservers = None         ,59                  pp        = False        ,60                  silent    = False        ,61                  progress  = True         , **kwargs ) :62        ##63        if isinstance ( ncpus , int ) and 1 <= ncpus : pass64        else                                         : ncpus = MP.cpu_count()65        if pp        :66            logger.warning ( "WorkManager: option ``pp'' is ignored" )67        if ppservers :68            logger.warning ( "WorkManager: option ``ppservers'' is ignored" )69        70        ## initialize the base class 71        TaskManager.__init__  ( self , ncpus = ncpus , silent = silent , progress = progress )        72        73        self.pool   = MP.Pool ( self.ncpus )74    # =========================================================================75    ## process the bare <code>executor</code> function76    #  @param job   function to be executed77    #  @param jobs_args the arguments, one entry per job 78    #  @return iterator to results 79    #  @code80    #  mgr  = WorkManager  ( .... )81    #  job  = ...82    #  args = ...83    #  for result in mgr.iexecute ( func , args ) :84    #  ...85    #  ... 86    #  @endcode87    #  It is a "bare minimal" interface88    #  - no statistics89    #  - no summary printout 90    #  - no merging of results  91    def iexecute ( self , job , jobs_args , progress = False ) :92        """Process the bare `executor` function93        >>> mgr  = WorkManager  ( .... )94        >>> job  = ...95        >>> args = ...96        >>> for result in mgr.iexecute ( job , args ) :97        ...98        ...99        It is a ``minimal'' interface100        - no statistics101        - no summary prin102        - no merging of results  103        """104                105        with pool_context ( self.pool ) as pool :106            ## create and submit jobs 107            jobs = pool.imap_unordered ( job , jobs_args )108            109            njobs  = len ( jobs_args ) if isinstance ( jobs_args , Sized ) else None            110            silent = self.silent or not progress111            112            ## retrive (asynchronous) results from the jobs113            for result in progress_bar ( jobs , max_value = njobs , silent = silent ) :114                yield result                115    # ========================================================================-116    ## get PP-statistics if/when posisble 117    def get_pp_stat ( self ) : 118        """Get PP-statistics if/when posisble 119        """120        return None 121            122# =============================================================================123if '__main__' == __name__ :124    125    from ostap.utils.docme import docme126    docme ( __name__ , logger = logger )    127    logger.info ("Module ``%s'' is used for multiprocessing" % MP.__name__ )128        129# =============================================================================130##                                                                      The END ...

Queue.py

Source:Queue.py

1from tactics.Singleton import *2import random345class Queue:6    unitmap = dict()7    def getUnitList(self):8        return self.unitmap.keys()9    def addToQueue(self, unit,action):10        11        action.running = True12        13        if not hasattr(unit, "timeleft"):14            unit.timeleft = 015        if not self.unitmap.has_key(unit):16            self.unitmap[unit] = [] 17        if hasattr(action, "reset"):18            action.reset()19        self.unitmap[unit].append(action)20#        if not unit in self.unitqueues:21#            self.unitqueues.append(unit)2223         24 #   unitqueues = []   25    def getActiveQueue(self):26        return len(self.unitmap)2728    def clearActions(self,unit):29        #dir([])30        if self.unitmap.has_key(unit):31            del self.unitmap[unit]32    def clearUnitQueue(self):33        self.unitmap = dict()34            35    def isActive(self,unit):36        37        if self.unitmap.has_key(unit):38            return True39        40                  41        return False42    a =       043    def runQueue(self,timer):44        45        self.a += timer46        for unit in self.unitmap.keys():47            48            49            if unit.timeleft > 0:50                unit.timeleft -= timer51                continue52           53            54            if len(self.unitmap[unit]) == 0:55                del self.unitmap[unit]56                continue57            iexecute = self.unitmap[unit][0]58            boo = False59           60            boo = iexecute.execute(timer)61            62            if not boo and len(self.unitmap[unit]):63                if hasattr(unit, "name"):64                    name = unit.name65                else:66                    name = str(unit)67                s.log("executable done "+ name+" "+str(iexecute),self)68                self.unitmap[unit].pop(0)69                if hasattr(iexecute, "timeleft"):70                    71                    unit.timeleft = iexecute.timeleft72                    unit.timeleft += random.random()*.173                    74               75                76                    77                    78                79            80#            if s.turnbased :   
...

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.