KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use run_once method in autotest

Best Python code snippet using autotest_python
gal.py
Source:gal.py
1"""2gal v0.013ga-bitbot application / system launcher4Copyright 2011 Brian Monkaba5This file is part of ga-bitbot.6    ga-bitbot is free software: you can redistribute it and/or modify7    it under the terms of the GNU General Public License as published by8    the Free Software Foundation, either version 3 of the License, or9    (at your option) any later version.10    ga-bitbot is distributed in the hope that it will be useful,11    but WITHOUT ANY WARRANTY; without even the implied warranty of12    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the13    GNU General Public License for more details.14    You should have received a copy of the GNU General Public License15    along with ga-bitbot.  If not, see <http://www.gnu.org/licenses/>.16"""17__appversion__ = "0.01a"18print "ga-bitbot application system launcher v%s"%__appversion__19import atexit20import sys21from subprocess import check_output as call, Popen, PIPE22import shlex23from os import environ24import os25from time import *26import hashlib27import random28import __main__29import paths30from load_config import *31import gene_server_config32import xmlrpclib33import json34import gc35def make_pid():36    #simple function which spits out a random hex code37    #which are used to set globaly unique process ids to spawned clients38    md = hashlib.md5()39    md.update(str(time()) + str(random.random() * 1000000))40    return md.hexdigest()[0:16]41print "-"*8042print "\n\tCommand line options:\n\t\tserver\t\tlaunches only the server components\n\t\tclient\t\tlaunches only the client components\n\t\tall\t\tlaunches all components"43print "\n\tThe default configuration is 'all'"44print "-"*8045run_client = 146run_server = 147mode = ""48if len(sys.argv) >= 2:49    if sys.argv[1] == 'all':50        mode = 'all'51        run_client = 152        run_server = 153        print "gal: launching all components"54    if sys.argv[1] == 'client':55        mode = 'client'56        run_client = 157        run_server = 058        print "gal: launching client components only"59    if sys.argv[1] == 'xlclient':60        mode = 'xlclient'61        run_client = 162        run_server = 063        print "gal: launching xlclient components only"64    if sys.argv[1] == 'server':65        mode = 'server'66        run_client = 067        run_server = 168        print "gal: launching server components only"69else:70    mode = 'all'71    print "gal: launching all components"72    sleep(3)73#the variable values below are superceded by the configuration loaded from the74#configuration file global_config.json75#!!!!!!!! to change the values edit the json configuration file NOT the variables below !!!!!!!!76WATCHDOG_TIMEOUT = 60 #seconds77MONITORED_PROCESS_LAUNCH_TIMEOUT = 20 #seconds78GENE_SERVER_STDERR_FILE = "/dev/null"79BCFEED_STDERR_FILE = "/dev/null"80WC_SERVER_STDERR_FILE = "/dev/null"81REPORT_GEN_STDERR_FILE = "/dev/null"82GTS_STDERR_FILE = "/dev/null"83config_loaded = False84#load config85try:86    __main__ = load_config_file_into_object('global_config.json',__main__)87except:88    print "gal: Error detected while loading the configuration. The application will now exit."89    import sys90    sys.exit()91else:92    if config_loaded == False:93        print "gal: Configuration failed to load. The application will now exit."94        import sys95        sys.exit()96    else:97        print "gal: Configuration loaded."98#open a null file to redirect stdout/stderr from the launched subprocesses99fnull = open(os.devnull,'w')100if GENE_SERVER_STDERR_FILE == "/dev/null":101    GENE_SERVER_STDERR_FILE = fnull102else:103    GENE_SERVER_STDERR_FILE = open(GENE_SERVER_STDERR_FILE,'a')104if BCFEED_STDERR_FILE == "/dev/null":105    BCFEED_STDERR_FILE = fnull106else:107    BCFEED_STDERR_FILE = open(BCFEED_STDERR_FILE,'a')108if WC_SERVER_STDERR_FILE == "/dev/null":109    WC_SERVER_STDERR_FILE = fnull110else:111    WC_SERVER_STDERR_FILE = open(WC_SERVER_STDERR_FILE,'a')112if REPORT_GEN_STDERR_FILE == "/dev/null":113    REPORT_GEN_STDERR_FILE = fnull114else:115    REPORT_GEN_STDERR_FILE = open(REPORT_GEN_STDERR_FILE,'a')116if GTS_STDERR_FILE == "/dev/null":117    GTS_STDERR_FILE = fnull118else:119    GTS_STDERR_FILE = open(GTS_STDERR_FILE,'a')120#configure gts clients based on the mode of operation (all,server or client)121#122# all - balanced123# server - focused on updating scores124# client - focused on finding new genes125#126# At least one gts instance in each mode should not run with the get_config option127# to make sure any new gene_def.json configs get loaded into the db.128#129all_monitored_launch = ['pypy gts.py all n run_once pid ',\130'pypy gts.py 3 n run_once get_config pid ',\131'pypy gts.py 3 y run_once get_config pid ',\132'pypy gts.py 4 n run_once get_config pid ',\133'pypy gts.py 4 y run_once get_config pid ',\134'pypy gts.py all y run_once get_config score_only pid ',\135'pypy gts.py all y run_once get_config pid ']136server_monitored_launch = ['pypy gts.py all y run_once pid ',\137'pypy gts.py 1 y run_once get_config score_only pid ',\138'pypy gts.py 2 y run_once get_config score_only pid ',\139'pypy gts.py 3 y run_once get_config score_only pid ',\140'pypy gts.py 3 y run_once get_config score_only pid ',\141'pypy gts.py 4 y run_once get_config score_only pid ',\142'pypy gts.py 4 y run_once get_config score_only pid ']143client_monitored_launch = ['pypy gts.py all n run_once pid ',\144'pypy gts.py 1 n run_once get_config pid ',\145'pypy gts.py 2 n run_once get_config pid ',\146'pypy gts.py 3 n run_once get_config pid ',\147'pypy gts.py 3 y run_once get_config pid ',\148'pypy gts.py 4 n run_once get_config pid ',\149'pypy gts.py 4 y run_once get_config pid ',\150'pypy gts.py all y run_once get_config pid ']151xlclient_monitored_launch = ['pypy gts.py all n run_once pid ',\152'pypy gts.py 1 n run_once get_config pid ',\153'pypy gts.py 2 n run_once get_config pid ',\154'pypy gts.py 3 n run_once get_config pid ',\155'pypy gts.py 3 n run_once get_config pid ',\156'pypy gts.py 4 n run_once get_config pid ',\157'pypy gts.py 4 n run_once get_config pid ',\158'pypy gts.py 3 y run_once get_config pid ',\159'pypy gts.py 3 y run_once get_config pid ',\160'pypy gts.py 4 y run_once get_config pid ',\161'pypy gts.py 4 y run_once get_config pid ',\162'pypy gts.py 1 n run_once get_config pid ',\163'pypy gts.py 2 n run_once get_config pid ',\164'pypy gts.py 3 n run_once get_config pid ',\165'pypy gts.py 3 n run_once get_config pid ',\166'pypy gts.py 4 n run_once get_config pid ',\167'pypy gts.py 4 n run_once get_config pid ',\168'pypy gts.py 3 y run_once get_config pid ',\169'pypy gts.py 3 y run_once get_config pid ',\170'pypy gts.py 4 y run_once get_config pid ',\171'pypy gts.py 4 y run_once get_config pid ',\172'pypy gts.py all y run_once get_config pid ']173if mode == 'all':174    monitored_launch = all_monitored_launch175if mode == 'server':176    monitored_launch = server_monitored_launch177if mode == 'client':178    monitored_launch = client_monitored_launch179if mode == 'xlclient':180    monitored_launch = xlclient_monitored_launch181unmonitored_launch = ['pypy wc_server.py','pypy report_gen.py']182monitor = {}    #variables to track monitored/unmonitored processes183no_monitor = []184def terminate_process(process):185    if sys.platform == 'win32':186        import ctypes187        PROCESS_TERMINATE = 1188        pid = process.pid189        handle = ctypes.windll.kernel32.OpenProcess(PROCESS_TERMINATE, False, pid)190        ctypes.windll.kernel32.TerminateProcess(handle, -1)191        ctypes.windll.kernel32.CloseHandle(handle)192    else:193        if process.poll() == None:194            process.terminate()195            process.wait()196# create and register callback function to do a clean shutdown of the system on exit.197def shutdown():198    global monitor199    global no_monitor200    global run_server201    for p in no_monitor:202        terminate_process(p)203    for pid in monitor.keys():204        terminate_process(monitor[pid]['process'])205    sys.stderr = fnull206    if run_server:207        server.shutdown()208atexit.register(shutdown)209#capture the price feeds regardless of client or server mode210#servers need it for reporting and clients need it for processing211#update the dataset212print "gal: Synching the local datafeed..."213Popen(shlex.split('python bcfeed_synch.py -d')).wait()214#launch the bcfeed script to collect data from the live feed215print "gal: Starting the live datafeed capture script..."216p = Popen(shlex.split('python bcfeed.py'),stdin=fnull, stdout=fnull, stderr=BCFEED_STDERR_FILE)217no_monitor.append(p)218if run_server:219    print "gal: Launching the xmlrpc server..."220    Popen(shlex.split('pypy gene_server.py'),stdin=fnull, stdout=fnull, stderr=GENE_SERVER_STDERR_FILE)221    sleep(5) #give the server time to start222# connect to the xml server223#224__server__ = gene_server_config.__server__225__port__ = str(gene_server_config.__port__)226server = xmlrpclib.Server('http://' + __server__ + ":" + __port__)227print "gal: connected to gene server ",__server__,":",__port__228if mode == 'all' or mode == 'server':229    print "gal: gene server db restore: ",server.reload()230print "gal: Launching GA Clients..."231#collect system process PIDS for monitoring.232#(not the same as system OS PIDs -- They are more like GUIDs as this is a multiclient distributed system)233epl = json.loads(server.pid_list()) #get the existing pid list234#start the monitored processes235for cmd_line in monitored_launch:236    new_pid = make_pid()237    p = Popen(shlex.split(cmd_line + new_pid),stdin=fnull, stdout=fnull, stderr=GTS_STDERR_FILE)238    retry = MONITORED_PROCESS_LAUNCH_TIMEOUT239    while retry > 0:240        sleep(1)241        cpl = json.loads(server.pid_list()) #get the current pid list242        npl = list(set(epl) ^ set(cpl))     #find the new pid(s)243        epl = cpl               #update the existing pid list244        if new_pid in npl:245            monitor.update({new_pid:{'cmd':cmd_line,'process':p}})  #store the pid/cmd_line/process246            print "gal: Monitored Process Launched (PID:",new_pid,"CMD:",cmd_line,")"247            break248        else:249            retry -= 1250    if retry == 0:251        print "gal: ERROR: Monitored Process Failed to Launch","(CMD:",cmd_line,")"252if run_server:253    #start unmonitored processes254    for cmd_line in unmonitored_launch:255        p = Popen(shlex.split(cmd_line),stdin=fnull, stdout=fnull, stderr=fnull)256        print "gal: Unmonitored Process Launched (CMD:",cmd_line,")"257        no_monitor.append(p)    #store the popen instance258        sleep(1) #wait a while before starting the report_gen script259print "\ngal: Monitoring Processes..."260count = 0261while 1:262    gc.collect()263    if run_server:264        count += 1265        #periodicaly tell the server to save the gene db266        if count == 50:267            count = 0268            server.save()269        if run_client == 0:270            sleep(30)271    #process monitor loop272    for pid in monitor.keys():273        sleep(5) #check one pid every n seconds.274        if server.pid_check(pid,WATCHDOG_TIMEOUT) == "NOK":275            #watchdog timed out276            print "gal: WATCHDOG: PID",pid,"EXPIRED"277            #remove the expired PID278            server.pid_remove(pid)279            epl = json.loads(server.pid_list()) #get the current pid list280            cmd_line = monitor[pid]['cmd']281            #terminate the process282            terminate_process(monitor[pid]['process'])283            monitor.pop(pid)284            #launch new process285            launching = 1286            while launching == 1:287                new_pid = make_pid()288                p = Popen(shlex.split(cmd_line + new_pid),stdin=fnull, stdout=fnull, stderr=GTS_STDERR_FILE)289                retry = MONITORED_PROCESS_LAUNCH_TIMEOUT290                while retry > 0:291                    sleep(1)292                    cpl = json.loads(server.pid_list()) #get the current pid list293                    npl = list(set(epl) ^ set(cpl))     #find the new pid(s)294                    epl = cpl               #update the existing pid list295                    if new_pid in npl:296                        monitor.update({new_pid:{'cmd':cmd_line,'process':p}})  #store the pid/cmd_line/process297                        print "gal: Monitored Process Launched (PID:",new_pid,"CMD:",cmd_line,")"298                        launching = 0299                        break300                    else:301                        retry -= 1302                if retry == 0:303                    print "gal: ERROR: Monitored Process Failed to Launch","(CMD:",cmd_line,")"...
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.