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

How to use doCode method in autotest

Best Python code snippet using autotest_python

cluster.py

Source:cluster.py

1"""2Cobra's built in clustering framework3"""4import sys5import time6import cobra7import dcode8import Queue9import struct10import socket11import threading12import subprocess13cluster_port = 3212314cluster_ip = "224.69.69.69"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        self.name = 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.name, 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        self.name = 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 (self.name != name) and (self.name != "*"):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        work.work()...

parse_qwasar.py

Source:parse_qwasar.py

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)11br.open("https://casapp.us.qwasar.io/login?service=https%3A%2F%2Fupskill.us.qwasar.io%2Fusers%2Fservice", timeout=10)12br.select_form(nr=0)13br.form['username'] = 'jas.884@mail.ru'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                br.open(f"https://upskill.us.qwasar.io/users/{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().day.at("03:00").do(main)87while True:88    schedule.run_pending()
...

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.