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

How to use startprocessmonitor method in pyatom

Best Python code snippet using pyatom_python

run.py

Source:run.py

1from pysys.constants import *2from pysys.basetest import BaseTest3from pysys.process.monitor import *4from pysys.utils.pycompat import openfile5if PROJECT.testRootDir+'/internal/utilities/extensions' not in sys.path:6	sys.path.append(PROJECT.testRootDir+'/internal/utilities/extensions') # only do this in internal testcases; normally sys.path should not be changed from within a PySys test7from pysysinternalhelpers import *8class PySysTest(BaseTest):9	def execute(self):10		p = self.startProcess(command=sys.executable,11						  arguments = [self.input+'/spinner.py'],12						  stdout = "%s/test.out" % self.output,13						  stderr = "%s/test.err" % self.output,14						  state=BACKGROUND)15						  16		childtest = runPySys(self, 'pysys', ['run', '-o', self.output+'/myoutdir', '-X', 'pidToMonitor=%d'%p.pid], 17			state=BACKGROUND, workingDir=self.input+'/nestedtests')18		# should be a no-op, just check it doesn't throw19		ProcessMonitorTextFileHandler.setDefaults(ProcessMonitorTextFileHandler.DEFAULT_COLUMNS)20		pm = self.startProcessMonitor(p, interval=0.1, file='monitor-default.tsv')21		pm2 = self.startProcessMonitor(p, interval=0.1, file=self.output+'/monitor-numproc.tsv', numProcessors='10')22		# test all supported stats, and also use of stream rather than filename23		filehandle = openfile(self.output+'/monitor-all.csv', 'w', encoding='utf-8')24		self.addCleanupFunction(lambda: filehandle.close())25		pm_all = self.startProcessMonitor(p, interval=0.1,handlers=[26			ProcessMonitorTextFileHandler(file=filehandle, columns=[27				ProcessMonitorKey.DATE_TIME,28				ProcessMonitorKey.SAMPLE,29				ProcessMonitorKey.CPU_CORE_UTILIZATION,30				ProcessMonitorKey.CPU_TOTAL_UTILIZATION,31				ProcessMonitorKey.MEMORY_RESIDENT_KB,32				ProcessMonitorKey.MEMORY_VIRTUAL_KB,33			], delimiter=',')34		])35		pidmonitor = self.startProcessMonitor(p.pid, interval=0.1, file=self.output+'/monitor-pid.tsv')36		assert pm.running(), 'monitor is still running'37		self.waitForGrep('monitor-default.tsv', expr='.', condition='>=5', ignores=['#.*'])38		self.waitForGrep('monitor-numproc.tsv', expr='.', condition='>=5', ignores=['#.*'])39		self.waitForGrep('monitor-pid.tsv', expr='.', condition='>=5', ignores=['#.*'])40		self.waitForGrep('monitor-all.csv', expr='.', condition='>=5', ignores=['#.*'])41		assert pm.running(), 'monitor is still running'42		assert pidmonitor.running(), 'pid monitor is still running'43		self.stopProcessMonitor(pidmonitor)44		self.waitProcess(childtest, timeout=60)45		self.assertTrue(childtest.exitStatus==0, assertMessage='nested pysys test passed')46		self.stopProcessMonitor(pm)47		pm.stop() # should silently do nothing48		self.stopProcessMonitor(pm) # should silently do nothing49		assert p.exitStatus == None, 'spinner process terminated unexpectedly'50		p.stop()51		self.wait(1) # keep process monitor running after it to check it doesn't cause an error52		self.stopProcessMonitor(pm2)53						  54		55	def validate(self):56		self.logFileContents('monitor-default.tsv')57		self.logFileContents('monitor-numproc.tsv')58		self.logFileContents('monitor-all.csv')59		self.logFileContents('myoutdir/NestedTest/monitor-legacy.tsv')60		with open(self.output+'/myoutdir/NestedTest/monitor-legacy.tsv') as f:61			header = f.readline()62			f.readline() # ignore first line of results63			line = f.readline().strip() 64		# ensure tab-delimited output has same number of items as header65		line = line.split('\t')66		self.log.info('Sample legacy log line:   %s', line)67		self.assertThat('%d == 4', len(line)) 68		self.assertGrep('myoutdir/NestedTest/monitor-legacy.tsv', expr='#.*', contains=False) # no header line69		self.log.info('')70		71		with open(self.output+'/monitor-default.tsv') as f:72			header = f.readline()73			self.assertTrue(header.startswith('#')) 74			f.readline() # ignore first line of results75			line = f.readline().strip() 76		# ensure tab-delimited output has same number of items as header77		line = line.split('\t')78		self.log.info('Sample log line:   %s', line)79		self.assertThat('%d == %d', len(line), len(header.split('\t'))) # same number of items in header line as normal lines80		for i in range(len(line)):81			if i > 0: # apart from the first column, every header should be a valid float or int82				try:83					float(line[i])84				except Exception:85					self.addOutcome(FAILED, 'monitor-default.tsv sample line [%d] is not a number: "%s"'%line[i])86		with open(self.output+'/monitor-all.csv') as f:87			header = f.readline()88			self.assertTrue(header.startswith('#')) 89			f.readline() # ignore first line of results90			line = f.readline().strip() 91		# ensure delimited output has same number of items as header92		self.assertGrep('monitor-all.csv', expr='\t', contains=False) # no tabs, since we use commas for this file93		line = line.split(',')94		self.log.info('Sample log line:   %s', line)95		self.assertThat('%d == %d', len(line), len(header.split(','))) # same number of items in header line as normal lines96		for i in range(len(line)):97			if i > 0: # apart from the first column, every header should be a valid float or int98				try:99					float(line[i])100				except Exception:101					self.addOutcome(FAILED, 'monitor-all.csv sample line [%d] is not a number: "%s"'%(i, line[i]))102		self.assertGrep('monitor-all.csv', expr='.*[.]', contains=False) # no floats, currently we expect all our stats to be integral103		104		# check files have at least some valid (non -1 ) values105		self.assertGrep('monitor-default.tsv', expr='\t[0-9]+')106		self.assertGrep('monitor-pid.tsv', expr='\t[0-9]+')107		self.assertGrep('myoutdir/NestedTest/monitor-legacy.tsv', expr='\t[0-9]+')108		109		# should be nothing where we couldn't get the data110		self.assertGrep('myoutdir/NestedTest/monitor-legacy.tsv', expr='\t-1', contains=False)...

pywindbg.py

Source:pywindbg.py

1# pywindbg.py2# Created by Starwarsfan2099 on 4/22/20193import argparse4from colorama import Fore5import commandParse6import utilities7version = "0.9"8utils = utilities.Utilities()9utils.dbgPrint("", Fore.GREEN)10utils.dbgPrint("PyWinDbg", Fore.GREEN)11utils.dbgPrint("Version %s" % version)12utils.dbgPrint("by Starwarsfan2099\n")13# Add all our command line arguments14parser = argparse.ArgumentParser()15parser.add_argument("-p", "--pid", type=int, help="process PID to attach to.")16parser.add_argument("-o", "--open", type=str, help="executable to launch and attach to.")17parser.add_argument("-l", "--log", type=str, help="log file to log to.")18parser.add_argument("-v", "--verbose", help="print more information.", action="store_true")19parser.add_argument("-d", "--debug", help="gives debugging information for development and bug hunting.", action="store_true")20parser.add_argument("-x", "--hide", help="hides debugger from debugee when first breakpoint is hit.", action="store_true")21parser.add_argument("-c", "--crash_mode", help="places hooks and prints information when a process crashes.", action="store_true")22parser.add_argument("-f", "--file_mode", help="hooks file creation and modification functions, printing info when called.", action="store_true")23parser.add_argument("-pm", "--process_monitor", help="starts monitoring all created processes.", action="store_true")24parser.add_argument("-fm", "--file_monitor", help="monitors file modification, creation, and deletion.", action="store_true")25args = parser.parse_args()26parser = commandParse.Parser()27# Parse the command line arguments given, if any28if args.log is not None:                                            # Log file29    parser.variables["logfile"] = args.log30    parser.variableParse("set logging True")31    utils.dbgPrint("[*] Logfile = %s" % args.log, Fore.GREEN)32if args.open is not None:                                           # Opening a file from command line33    parser.openAndAttach(args.open)34if args.verbose is True:                                            # Verbose mode35    parser.variables["verbose"] = True36    parser.variableParse("set verbose True")37    utils.dbgPrint("[*] Verbose = True", Fore.GREEN)38if args.debug is True:                                              # Debug mode39    parser.variables["debug"] = True40    parser.variableParse("set debug True")41    utils.dbgPrint("[*] Debug = True", Fore.GREEN)42if args.hide is True:                                               # Hide debugger43    parser.variables["hide-debugger"] = True44    parser.variableParse("set hide-debugger True")45    utils.dbgPrint("[*] Hide-debugger = True", Fore.GREEN)46if args.crash_mode is True:                                         # Crash Mode47    parser.variables["crash-mode"] = True48    parser.variableParse("set crash-mode True")49    utils.dbgPrint("[*] Crash-mode = True", Fore.GREEN)50if args.file_mode is True:                                          # File Mode51    parser.variables["file-mode"] = True52    parser.variableParse("set file-mode True")53    utils.dbgPrint("[*] File-mode = True", Fore.GREEN)54if args.process_monitor is True:                                    # Process monitor55    parser.startProcessMonitor()56if args.file_monitor is True:                                       # File monitor57    parser.startFileMonitor("fm")58utils.dbgPrint("")59# Start the main command prompt loop60if __name__ == '__main__':61    while True:...

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.