How to use doCode method in autotest

Best Python code snippet using autotest_python Github


Full Screen

1"""2Cobra's built in clustering framework3"""4import sys5import time6import cobra7import dcode8import Queue9import struct10import socket11import threading12import subprocess13cluster_port = 3212314cluster_ip = ""15sub_cmd = """16import cobra.cluster17import cobra.dcode18import urllib219if %s:20 x = urllib2.Request("%s")21 cobra.dcode.enableDcodeClient()22 cobra.dcode.addDcodeServer(x.get_host().split(":")[0])23cobra.cluster.getAndDoWork("%s")24"""25class ClusterWork(object):26 """27 Extend this object to create your own work units. Do it in28 a proper module (and not __main__ to be able to use this29 in conjunction with cobra.dcode).30 """31 def __init__(self):32 object.__init__(self)33 def work(self):34 """35 Actually do the work associated with this work object.36 """37 print "OVERRIDE ME"38 def done(self):39 """40 This is called back on the server once a work unit41 is complete and returned.42 """43 print "OVERRIDE DONE"44class ClusterServer:45 def __init__(self, name, maxsize=0, docode=False):46 self.added = False47 = name48 self.inprog = 049 self.maxsize = maxsize50 self.queue = Queue.Queue(maxsize)51 self.cobraname = cobra.shareObject(self)52 if docode: dcode.enableDcodeServer()53 def runServer(self):54 sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)55 while (self.added == False or56 self.queue.empty() == False or57 self.inprog > 0):58 buf = "cobra:%s:%s:%d" % (, self.cobraname, cobra.COBRA_PORT)59 sock.sendto(buf, (cluster_ip, cluster_port))60 time.sleep(1)61 def addWork(self, work):62 """63 Add a work object to the ClusterServer. This 64 """65 self.added = True # One time add detection66 if not isinstance(work, ClusterWork):67 raise Exception("%s is not a ClusterWork extension!")68 self.queue.put(work)69 def getWork(self):70 try:71 ret = self.queue.get_nowait()72 self.inprog += 173 return ret74 except Queue.Empty, e:75 return None76 def doneWork(self, work):77 """78 Used by the clients to report work as done.79 """80 self.inprog -= 181 work.done()82class ClusterClient:83 """84 Listen for our name (or any name if name=="*") on the cobra cluster85 multicast address and if we find a server in need, go help.86 maxwidth is the number of work units to do in parallel87 docode will enable code sharing with the server88 threaded == True will use threads, otherwise subprocess of the python interpreter (OMG CLUSTER)89 """90 def __init__(self, name, maxwidth=4, threaded=True, docode=False):91 self.go = True92 = name93 self.width = 094 self.maxwidth = maxwidth95 self.threaded = threaded96 self.verbose = False97 self.docode = docode98 if docode: dcode.enableDcodeClient()99 self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)100 self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)101 self.sock.bind(("",cluster_port))102 mreq = struct.pack("4sL", socket.inet_aton(cluster_ip), socket.INADDR_ANY)103 self.sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)104 def processWork(self):105 """106 Runs handing out work up to maxwidth until self.go == False.107 """108 while self.go:109 110 buf, sockaddr = self.sock.recvfrom(4096)111 if self.width >= self.maxwidth:112 continue113 if not buf.startswith("cobra:") and not buf.startswith("cobrassl:"):114 continue115 info = buf.split(":")116 if len(info) != 4:117 continue118 server, svrport = sockaddr119 cc,name,cobject,portstr = info120 if ( != name) and ( != "*"):121 continue122 port = int(portstr)123 #FIXME this should fire a thread...124 if self.docode:125 dcode.addDcodeServer(server, port=port)126 uri = "%s://%s:%d/%s" % (cc,server,port,cobject)127 self.fireRunner(uri)128 def fireRunner(self, uri):129 if self.threaded:130 thr = threading.Thread(target=self.threadWorker, args=(uri,))131 thr.setDaemon(True)132 thr.start()133 else:134 thr = threading.Thread(target=self.threadForker, args=(uri,))135 thr.setDaemon(True)136 thr.start()137 def threadWorker(self, uri):138 self.width += 1139 try:140 return getAndDoWork(uri)141 finally:142 self.width -= 1143 def threadForker(self, uri):144 self.width += 1145 cmd = sub_cmd % (self.docode, uri, uri)146 try:147 sub = subprocess.Popen([sys.executable, '-c', cmd])148 sub.wait()149 finally:150 self.width -= 1151def getAndDoWork(uri):152 proxy = cobra.CobraProxy(uri)153 work = proxy.getWork()154 # If we got work, do it.155 if work != None:156

Full Screen

Full Screen Github


Full Screen

1from bs4 import BeautifulSoup as bs2import pandas as pd3import mechanize4import http.cookiejar as cookielib5import schedule6import time78cj = cookielib.CookieJar()9br = mechanize.Browser()10br.set_cookiejar(cj)"", timeout=10)12br.select_form(nr=0)13br.form['username'] = ''14br.form['password'] = 'jasur2171517'15br.submit()161718def main():19 def get_data_from_qwasar(data, usernames):20 student_data = {}21 counter = 022 for i in usernames:23 try:24"{i}")25 soup = bs(br.response().read(), 'html.parser')2627 student_data['qwasar username'] = i28 student_data['ism-familiya'] = soup.find('h1', class_='text-c-yellow').text.strip()29 student_data['qwasardagi ballari '] = soup.find('div', {'id': 'user_qpoints'}).text.strip()30 student_data["hozirgi vaqtdagi o'qiyotgan sezoni"] = soup.find('div',31 class_='row p-t-10 align-items-center').find(32 'a').text.strip()33 student_data['qwasardagi hozirgi foizi'] = soup.find("div", class_="progress b-radius-1")[34 'title'].strip()3536 months = {'January': 'Yanvar', 'February': 'Fevral', 'March': 'Mart', 'April': 'Aprel', 'May': 'May',37 'June': 'Iyun', 'July': 'Iyul', 'August': 'Avgust', 'September': 'Sentabr',38 'October': 'Oktabr',39 'November': 'Noyabr', 'December': 'Dekabr'}4041 student_months = soup.find_all("div", {"class": 'border col-sm-4'})4243 for month in student_months:44 sum_docode = 045 sum_upskill = 046 count_docode = 047 count_upskill = 048 student_active = month.find_all("div", {"class": 'c-pointer'}) # col text-c-white49 month_name = month.find_all("div", {"class": 'col text-c-white'})[0].text.strip()50 for link in student_active:51 try:52 if link['title']:53 docode = float(link['title'].split("<br>")[1][8:-1])54 upskill = float(link['title'].split("<br>")[2][8:-1])55 if docode > 0:56 count_docode += 157 if upskill > 0:58 count_upskill += 159 sum_docode += docode60 sum_upskill += upskill61 except:62 pass6364 student_data[f"{months[month_name]} oyida docode vaqti"] = sum_docode65 student_data[f"{months[month_name]} oyida platformadagi vaqti"] = sum_upskill66 student_data[f"{months[month_name]} oyida aktiv docode kunlari"] = count_docode67 student_data[f"{months[month_name]} oyida platformadagi aktiv kunlari"] = count_upskill68 except:69 pass7071 user_data = pd.DataFrame(pd.Series({i: str(student_data)[1:-1]}), columns=['data'])72 data = pd.concat([data, user_data])7374 print(counter, i)75 counter += 17677 return data7879 from database.students import usernames8081 full_data = get_data_from_qwasar(pd.DataFrame({}), usernames).reset_index()82 full_data.columns = ['usernames', 'data']83 full_data.to_csv('./database/real_database.csv', index=False)848586schedule.every()"03:00").do(main)87while True:88 schedule.run_pending() ...

Full Screen

Full Screen

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:


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

Next-Gen App & Browser Testing Cloud

Was this article helpful?