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

How to use by_pid method in lisa

Best Python code snippet using lisa_python

chrome_killa.py

Source:chrome_killa.py

1"""Kill runaway Chrome processes."""2import logging3import sys4import threading5import time6import psutil7from enum import Enum8CHROME_KILL_AFTER_S = 1809logger = logging.getLogger("killa")10class KillState(Enum):11    ALIVE = 012    TERMED = 113    KILLED = 214class WatchedProc(object):15    proc = None16    first_seen = None17    state = None18    def __init__(self, proc, first_seen, state):19        self.proc = proc20        self.first_seen = first_seen21        self.state = state22    def try_stop(self):23        if self.state == KillState.ALIVE:24            logger.info("TERMing runaway Chrome %d", self.proc.pid)25            self.proc.terminate()26            self.state = KillState.TERMED27        elif self.state == KillState.TERMED:28            logger.warning("KILLing runaway Chrome %d", self.proc.pid)29            self.proc.kill()30            self.state = KillState.KILLED31        elif (self.state == KillState.KILLED and self.proc.is_running() and32              self.proc.status() == 'running'):33            logger.error("Couldn't kill Chrome %d", self.proc.pid)34def _find_chromes():35    chromes = []36    #for proc in psutil.process_iter():  # This finds all instances of Chrome37    for proc in psutil.Process().children(True):  # Only our children38        try:39            if proc.name() == 'Google Chrome':40                chromes.append(proc)41        except psutil.ZombieProcess:42            pass43        except psutil.NoSuchProcess:44            pass45        except psutil.AccessDenied:46            pass47        except Exception:48            logger.exception("unknown exception caught")49            pass50    return chromes51def _watch_chromes():52    logger.info("Monitoring for long-term Chrome sessions.")53    by_pid = {}54    while True:55        cur_chromes = _find_chromes()56        now = time.time()57        for chrome in cur_chromes:58            if chrome.pid in by_pid:59                continue60            by_pid[chrome.pid] = WatchedProc(chrome, now, KillState.ALIVE)61            logger.debug("Monitoring new Chrome instance %d", chrome.pid)62        to_rm = []63        for pid, wproc in by_pid.items():64            try:65                if (not wproc.proc.is_running() or66                        wproc.proc.status() == 'zombie'):67                    to_rm.append(wproc.proc.pid)68                elif now - wproc.first_seen > CHROME_KILL_AFTER_S:69                    wproc.try_stop()70            except psutil.NoSuchProcess:71                to_rm.append(wproc.proc.pid)72            except ProcessLookupError:73                to_rm.append(wproc.proc.pid)74            except Exception:75                logger.error("Unexpected exception")76        for pid in to_rm:77            del by_pid[pid]78        time.sleep(2)79def start_chrome_killa():80    thread = threading.Thread(target=_watch_chromes)81    thread.daemon = True82    thread.start()83def main():84    start_chrome_killa()85    from selenium import webdriver86    webdriver.Chrome(87        "/Users/vpn_test/vpn_tests/manipulation_tests/"88        "redirection_dom/chromedriver")89    time.sleep(200)90if __name__ == "__main__":...

process.py

Source:process.py

1import os2import signal3class Process(object):4	"""Represents a process"""5	def __init__(self, pid):6		"""Make a new Process object"""7		self.proc = "/proc/%d" % pid8		f = open(os.path.join(self.proc, "stat"))9		pid,command,state,parent_pid = f.read().strip().split()[:4]10		f.close()11		command = command[1:-1]12		self.pid = int(pid)13		self.command = command14		self.state = state15		try:16			self.parent_pid = int(parent_pid)17		except:18			self.parent_pid = int(0)19		self.parent = None20		self.children = []21	def kill(self, sig = signal.SIGTERM):22		"""Kill this process with SIGTERM by default"""23		os.kill(self.pid, sig)24	def __repr__(self):25		return "%r" % self.pid26	def getcwd(self):27		"""Read the current directory of this process or None for can't"""28		try:29			return os.readlink(os.path.join(self.proc, "cwd"))30		except OSError:31			return None32class ProcessList(object):33	"""Represents a list of processes"""34	def __init__(self):35		"""Read /proc and fill up the process lists"""36		self.by_pid = {}37		self.by_command = {}38		for f in os.listdir("/proc"):39			try:40				if f.isdigit():41					process = Process(int(f))42					self.by_pid[process.pid] = process43					self.by_command.setdefault(process.command, []).append(process)44			except IOError:45				pass46		for process in self.by_pid.values():47			try:48				parent = self.by_pid[process.parent_pid]49				#print "child",process50				#print "parent",parent51				parent.children.append(process)52				process.parent = parent53			except KeyError:54				pass55	def named(self, name):56		"""Returns a list of processes with the given name"""...

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.