How to use find_pid method in autotest

Best Python code snippet using autotest_python

ShellCommand_Windows.py

Source:ShellCommand_Windows.py

1# coding=utf-82import os3import re4import subprocess5class ShellCommandWindows(object):6    """7    API8    The shell command of Windows.9    """10    def __init__(self):11        pass12    def kill_zombie_proc(self):13        """14        Kill zombie process.15        Avoid system resource crashes.16        nothing now.17        """18        pass19    def kill_other_python(self):20        """21        kill other python before launch this auto test tool.22        maybe some conflict23        """24        pass25        # port = re.findall(r"python.exe.+?(\d+).+", os.popen("tasklist|findstr python").read())26        # for i in [i for i in set(port) if str(os.getpid()) != i]:27        #     os.system("taskkill /f /t /pid %s" % i)28    def find_proc_and_pid_by_port(self, port):29        """30        find process and pid by process port for Windows.31        int -> list32        :return:[(proc1, pid1), (proc2, pid2)] [str, int]33        """34        try:35            int(port)36        except ValueError:37            raise KeyError("key must be port! Is int, but real %s!" % type(port))38        except TypeError:39            raise KeyError("key must be port! Is int, but real %s!" % type(port))40        command = 'netstat -aon|findstr %s' % port  # å¤æç«¯å£æ¯å¦è¢«å ç¨41        bind_pid = list(set(re.findall(r".+LISTEN.+?(\d+)", os.popen(command).read())))42        if not bind_pid:43            return bind_pid44        find_pid = []45        for i in bind_pid:46            command = 'tasklist|findstr %s' % i47            find_pid.append(re.findall(r"(.+?) .+?(\d+).+?Console.+", os.popen(command).read()))48        find_pid = sum(find_pid, [])  # éå½åè¡¨[[(),()],[()]] â [(),(),()]49        find_pid = [(i[0], int(i[1])) for i in find_pid if i[1] in bind_pid]50        return find_pid51    def find_proc_and_pid_by_pid(self, pid):52        """53        find process and pid by process pid for Windows.54        int -> list55        :return:[(proc1, pid1), (proc2, pid2)] [str, int]56        """57        try:58            int(pid)59        except ValueError:60            raise KeyError("key must be pid! Is int, but real %s!" % type(pid))61        except TypeError:62            raise KeyError("key must be pid! Is int, but real %s!" % type(pid))63        command = 'tasklist|findstr %s' % pid64        find_pid = list(set(re.findall(r"(.+?) .+?(\d+).+?Console.+", os.popen(command).read())))65        find_pid = [(i[0], int(i[1])) for i in find_pid if i[1] == pid]66        return find_pid67    def find_proc_and_pid_by_proc(self, proc):68        """69        find process and pid by process name for Windows.70        str -> list71        :return:[(proc1, pid1), (proc2, pid2)] [str, int]72        """73        if not isinstance(proc, str):74            raise KeyError("key must be process name! Is string, but real %s!" % type(proc))75        command = 'tasklist|findstr %s' % proc76        find_pid = list(set(re.findall(r"(.+?) .+?(\d+).+?Console.+", os.popen(command).read())))77        find_pid = list(map(lambda x: (x[0], int(x[1])), find_pid))78        return find_pid79    def kill_proc_by_proc(self, proc):80        """81        kill process by process name for Windows.82        """83        if not isinstance(proc, str):84            raise KeyError("key must be process name! Is string, but real %s!" % type(proc))85        if ".exe" in proc:86            command = 'taskkill /f /t /im %s' % proc  # éè¿è¿ç¨åææ»è¿ç¨87        else:88            command = 'taskkill /f /t /im %s.exe' % proc  # éè¿è¿ç¨åææ»è¿ç¨89        print(subprocess.check_output(command).decode("gbk"))90        if self.find_proc_and_pid_by_proc(proc):91            raise AssertionError("kill %s fail." % proc)92        else:93            print(u"ç»æ¢ %s è¿ç¨ã" % proc)94    def kill_proc_by_pid(self, pid):95        """96        kill process by process pid for Windows.97        """98        try:99            int(pid)100        except ValueError:101            raise KeyError("key must be pid! Is int, but real %s!" % type(pid))102        except TypeError:103            raise KeyError("key must be pid! Is int, but real %s!" % type(pid))104        command = 'taskkill /f /t /pid %s' % pid  # éè¿pidææ»è¿ç¨105        print(subprocess.check_output(command).decode("gbk"))106        # os.system(command)107        if self.find_proc_and_pid_by_pid(pid):108            raise AssertionError("kill %s fail." % pid)109        else:110            print(u"ç»æ¢ PID %sã" % pid)111    def set_appium_log_addr(self):112        """113        set appium server log repository path for Windows.114        """115        # addr = os.getenv("Temp")116        addr = os.getcwd()117        addr = os.path.join(addr, "debug")118        return addr119    def push_appium_app(self, udid):120        """121        ä»£æ¿appiumå®è£ä¸å¤§APP122        """123        command = r"adb -s %s install .\apk\unlock_apk-debug.apk" % udid124        os.system(command)125        command = r"adb -s %s install .\apk\settings_apk-debug.apk" % udid126        os.system(command)127        command = r"adb -s %s install .\apk\UnicodeIME-debug.apk" % udid128        os.system(command)129    def replace_appium_js(self):130        """131        appiumæ¯æ¬¡é½ä¼å®è£setting.apkåunlock.apkï¼å¤å¶å·²ç»åæ¶å®è£çä»£ç è³æºappiumè·¯å¾132        """133        appium_path = os.popen("where appium").read().split("\n")[0].split(".bin")[0]134        for i in ["android", "android-common", "adb"]:135            if i == "adb":136                tmp = os.path.join(appium_path, r"appium\node_modules\appium-adb\lib")137            else:138                tmp = os.path.join(appium_path, r"appium\lib\devices\android")139            old_path = os.path.join(tmp, "%s.js" % i)140            try:141                os.renames(old_path, os.path.join(tmp, "%s.bak" % i))142            except WindowsError:143                pass144            if not os.path.isfile(old_path):...

ShellCommand_Mac.py

Source:ShellCommand_Mac.py

1# coding=utf-82import os3import re4import subprocess5class ShellCommandMac(object):6    """7    API8    The shell command of Mac.9    """10    def __init__(self):11        pass12    def kill_zombie_proc(self):13        """14        Kill zombie process.15        Avoid system resource crashes.16        such idevicesyslog and mdworker.17        """18        # ä½¿ç¨popenæ§å¶å°ä¼æå¤§é No matching processes belonging to you were found19        os.popen("killall -9 idevicesyslog")20        os.popen("killall -9 mdworker")21    def kill_other_python(self):22        """23        kill other python before launch this auto test tool for Mac.24        maybe some conflict25        """26        port = re.findall(r"Python.+?(\d+) .+", os.popen("lsof -c Python").read())27        for i in [i for i in set(port) if str(os.getpid()) != i]:28            os.system("kill -9 %s" % i)29        self.kill_zombie_proc()30    def find_proc_and_pid_by_port(self, port):31        """32        find process and pid by process port for Mac.33        int -> list34        :return:[(proc1, pid1), (proc2, pid2)] [str, int]35        """36        try:37            int(port)38        except ValueError:39            raise KeyError("key must be port! Is int, but real %s!" % type(port))40        except TypeError:41            raise KeyError("key must be port! Is int, but real %s!" % type(port))42        command = 'lsof -i:%s' % port  # å¤æç«¯å£æ¯å¦è¢«å ç¨43        find_pid = list(set(re.findall(r"(.+?) .+?(\d+).+\(LISTEN.+?", os.popen(command).read())))44        find_pid = list(map(lambda x: (x[0], int(x[1])), find_pid))45        self.kill_zombie_proc()46        return find_pid47    def find_proc_and_pid_by_pid(self, pid):48        """49        find process and pid by process pid for Mac.50        int -> list51        :return:[(proc1, pid1), (proc2, pid2)] [str, int]52        """53        try:54            int(pid)55        except ValueError:56            raise KeyError("key must be pid! Is int, but real %s!" % type(pid))57        except TypeError:58            raise KeyError("key must be pid! Is int, but real %s!" % type(pid))59        command = 'lsof -p %s' % pid60        find_pid = list(set(re.findall(r"(.+?) .+?(\d+).+", os.popen(command).read())))61        find_pid = list(map(lambda x: (x[0], int(x[1])), find_pid))62        self.kill_zombie_proc()63        return find_pid64    def find_proc_and_pid_by_proc(self, proc):65        """66        find process and pid by process name for Mac.67        str -> list68        :return:[(proc1, pid1), (proc2, pid2)] [str, int]69        """70        command = 'lsof -c %s' % proc71        find_pid = list(set(re.findall(r"(.+?) .+?(\d+).+", os.popen(command).read())))72        find_pid = list(map(lambda x: (x[0], int(x[1])), find_pid))73        self.kill_zombie_proc()74        return find_pid75    def kill_proc_by_proc(self, proc):76        """77        kill process by process name for Mac.78        """79        if not isinstance(proc, str):80            raise KeyError("key must be process name! Is string, but real %s!" % type(proc))81        command = 'killall -9 %s' % proc  # éè¿è¿ç¨åææ»è¿ç¨82        os.system(command)83        self.kill_zombie_proc()84        if self.find_proc_and_pid_by_proc(proc):85            raise AssertionError("kill %s fail." % proc)86        else:87            print(u"ç»æ¢ %s è¿ç¨ã" % proc)88    def kill_proc_by_pid(self, pid):89        """90        kill process by process pid for Mac.91        """92        try:93            int(pid)94        except ValueError:95            raise KeyError("key must be pid! Is int, but real %s!" % type(pid))96        except TypeError:97            raise KeyError("key must be pid! Is int, but real %s!" % type(pid))98        command = 'kill -9 %s' % pid  # éè¿pidææ»è¿ç¨99        os.system(command)100        self.kill_zombie_proc()101        if self.find_proc_and_pid_by_pid(pid):102            raise AssertionError("kill %s fail." % pid)103        else:104            print(u"ç»æ¢ PID %sã" % pid)105    def set_appium_log_addr(self):106        """107        set appium server log repository path for Mac.108        """109        code = ["utf-8", "gbk"]110        addr = None111        for i in code:112            try:113                addr = os.getcwd().decode(i)114                addr = os.path.join(addr, "debug")115                break116            except UnicodeDecodeError:117                pass118        if addr is None:119            raise UnicodeDecodeError("%s codec can't decode bytes in os.getcwd()" % code)120        else:121            return addr122    def push_appium_app(self, udid):123        """124        ä»£æ¿appiumå®è£ä¸å¤§APP125        """126        command = r"adb -s %s install .\apk\unlock_apk-debug.apk" % udid127        os.system(command)128        command = r"adb -s %s install .\apk\settings_apk-debug.apk" % udid129        os.system(command)130        command = r"adb -s %s install .\apk\UnicodeIME-debug.apk" % udid131        os.system(command)132    def replace_appium_js(self):133        """134        appiumæ¯æ¬¡é½ä¼å®è£setting.apkåunlock.apkï¼å¤å¶å·²ç»åæ¶å®è£çä»£ç è³æºappiumè·¯å¾135        """136        appium_path = os.popen("where appium").read().split("\n")[0].split(".bin")[0]137        appium_js_path = os.path.join(appium_path, r"appium\lib\devices\android")138        for i in ["android", "android-common", "adb"]:139            if i == "adb":140                tmp = os.path.join(appium_path, r"appium\node_modules\appium-adb\lib")141            else:142                tmp = os.path.join(appium_path, r"appium\lib\devices\android")143            old_path = os.path.join(tmp, "%s.js" % i)144            try:145                os.renames(old_path, os.path.join(tmp, "%s.bak" % i))146            except WindowsError:147                pass148            if not os.path.isfile(old_path):...

manager.py

Source:manager.py

...18                 help="[optional] process name to start or stop specific process,"19                      " if not precent, mean start or stop all process")20def check_requirements():21    error_message = []22    postgres = find_pid('postgresql')23    if not postgres:24        error_message.append('postgresql service required, but not running!')25    return True if not error_message else error_message26def status():27    for process in constants.APP_PROCESSES:28        pids = find_pid(process.get('token'))29        if not pids:30            ApiLogging.error([process.get('name'), pids], True)31        else:32            ApiLogging.info([process.get('name'), pids], True)33def start(process_name=None):34    requirements = check_requirements()35    if requirements is not True:36        for requirement in requirements:37            ApiLogging.critical(requirement, True)38        return39    if process_name and __get_process(process_name) is not None:40        process = __get_process(process_name)41        pids = find_pid(process.get('token'))42        if pids:43            ApiLogging.warning(str(len(pids)) + ' instance(s) of this process already running!', True)44        else:45            __run(process_name, 'start')46    else:47        for process in constants.APP_PROCESSES:48            if find_pid(process.get('token')):49                ApiLogging.warning(process.get('name') + ' is already running!', True)50            else:51                __run(process.get('name'), 'start')52def stop(process_name=None):53    if process_name and __get_process(process_name) is not None:54        process = __get_process(process_name)55        pids = find_pid(process.get('token'))56        if not pids:57            ApiLogging.warning(process_name + ' is not running!', True)58        else:59            for pid in pids:60                ApiLogging.info(process_name + ' stopped successful!', True)61                proc = psutil.Process(pid)62                if proc.is_running():63                    proc.kill()64    else:65        for process in constants.APP_PROCESSES:66            pids = find_pid(process.get('token'))67            if not pids:68                ApiLogging.warning(process.get('name') + ' is not running!', True)69            else:70                for pid in pids:71                    ApiLogging.info(process.get('name') + ' stopped successful!', True)72                    proc = psutil.Process(pid)73                    if proc.is_running():74                        proc.kill()75    try:76        __cleanup()77    except Exception as e:78        # print("cleanup exception: ", e)79        pass80def __get_process(process_name):81    for process in constants.APP_PROCESSES:82        if process.get('name') == process_name:83            return process84    return None85def __set_limit():86    resource.setrlimit(resource.RLIMIT_NOFILE, (99999, 99999))87def __run(process_name, action_name):88    process = __get_process(process_name)89    if process:90        ApiLogging.info(process.get('name') + ' running successful!', True)91        with open(process.get('log'), 'a+') as err:92            subprocess.Popen(['python3', process.get('path') + '/' + process.get('name'), process.get('token')],93                             close_fds=True, stderr=err,bufsize=1, preexec_fn=__set_limit)94def __cleanup():95    # terminate all xvfb process96    pids = find_pid('Xvfb')97    for pid in pids:98        proc = psutil.Process(pid)99        if proc.is_running():100            proc.kill()101def set_mode(debug_mode):102    path = BASE_APP_PATH + '/components'103    file_name = 'mode.py'104    file_adress = os.path.join(path, file_name)105    f = open(file_adress, "w+")106    f.write('debug_mode = ' + str(debug_mode))107if not os.path.exists(BASE_APP_PATH + '/logs'):108    os.mkdir(BASE_APP_PATH + '/logs')109args = vars(arg.parse_args())110action = args.get('action')...

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.