KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use kill method in Airtest
Best Python code snippet using Airtest
test_async_process.py
Source:test_async_process.py 
...164 return_value=pid),\165 mock.patch.object(self.proc, '_kill_process_and_wait'166 ) as mock_kill_process_and_wait,\167 mock.patch.object(self.proc, '_process'):168 self.proc._kill(signal.SIGKILL)169 self.assertIsNone(self.proc._kill_event)170 self.assertFalse(self.proc._is_running)171 self.assertIsNone(self.proc._pid)172 mock_kill_event.send.assert_called_once_with()173 if pid:174 mock_kill_process_and_wait.assert_called_once_with(175 pid, signal.SIGKILL, None)176 def _test__kill_process_and_wait(self, pid, expected,177 exception_message=None,178 kill_signal=signal.SIGKILL):179 self.proc.run_as_root = True180 if exception_message:181 exc = RuntimeError(exception_message)182 else:...
killBufferCommands.py
Source:killBufferCommands.py
...57 return58 undoType = 'backward-kill-sentence'59 self.beginCommand(w, undoType=undoType)60 i2 = s.rfind('.', 0, i) + 161 self.kill(event, i2, i + 1, undoType=undoType)62 w.setInsertPoint(i2)63 self.endCommand(changed=True, setLabel=True)64 #@+node:ekr.20150514063305.413: *3* backwardKillWord & killWord65 @cmd('backward-kill-word')66 def backwardKillWord(self, event):67 """Kill the previous word."""68 c = self.c69 w = self.editWidget(event)70 if w:71 self.beginCommand(w, undoType='backward-kill-word')72 c.editCommands.backwardWord(event)73 self.killWordHelper(event)74 @cmd('kill-word')75 def killWord(self, event):76 """Kill the word containing the cursor."""77 w = self.editWidget(event)78 if w:79 self.beginCommand(w, undoType='kill-word')80 self.killWordHelper(event)81 def killWordHelper(self, event):82 c = self.c83 e = c.editCommands84 w = e.editWidget(event)85 if w:86 # self.killWs(event)87 e.extendToWord(event)88 i, j = w.getSelectionRange()89 self.kill(event, i, j, undoType=None)90 self.endCommand(changed=True, setLabel=True)91 #@+node:ekr.20150514063305.414: *3* clearKillRing92 @cmd('clear-kill-ring')93 def clearKillRing(self, event=None):94 """Clear the kill ring."""95 g.app.globalKillbuffer = []96 #@+node:ekr.20150514063305.415: *3* getClipboard97 def getClipboard(self):98 """Return the contents of the clipboard."""99 try:100 ctxt = g.app.gui.getTextFromClipboard()101 if not g.app.globalKillBuffer or ctxt != self.last_clipboard:102 self.last_clipboard = ctxt103 if not g.app.globalKillBuffer or g.app.globalKillBuffer[0] != ctxt:104 return ctxt105 except Exception:106 g.es_exception()107 return None108 #@+node:ekr.20150514063305.416: *3* class iterateKillBuffer109 class KillBufferIterClass:110 """Returns a list of positions in a subtree, possibly including the root of the subtree."""111 #@+others112 #@+node:ekr.20150514063305.417: *4* __init__ & __iter__ (iterateKillBuffer)113 def __init__(self, c):114 """Ctor for KillBufferIterClass class."""115 self.c = c116 self.index = 0 # The index of the next item to be returned.117 def __iter__(self):118 return self119 #@+node:ekr.20150514063305.418: *4* __next__120 def __next__(self):121 commands = self.c.killBufferCommands122 aList = g.app.globalKillBuffer # commands.killBuffer123 if not aList:124 self.index = 0125 return None126 if commands.reset is None:127 i = self.index128 else:129 i = commands.reset130 commands.reset = None131 if i < 0 or i >= len(aList): i = 0132 val = aList[i]133 self.index = i + 1134 return val135 #@-others136 def iterateKillBuffer(self):137 return self.KillBufferIterClass(self.c)138 #@+node:ekr.20150514063305.419: *3* kill (helper)139 def kill(self, event, frm, to, force=False, undoType=None):140 """A helper method for all kill commands."""141 w = self.editWidget(event)142 if not w: return143 # 2016/03/05: all kill commands kill selected text, if it exists.144 if not force:145 # Delete the selection range if it spans a line.146 i, j = w.getSelectionRange()147 s = w.get(i, j)148 if s.find('\n') > -1:149 frm, to = i, j150 s = w.get(frm, to)151 if undoType:152 self.beginCommand(w, undoType=undoType)153 self.addToKillBuffer(s)154 g.app.gui.replaceClipboardWith(s)155 w.delete(frm, to)156 w.setInsertPoint(frm)157 if undoType:158 self.endCommand(changed=True, setLabel=True)159 #@+node:ekr.20150514063305.420: *3* killToEndOfLine160 @cmd('kill-to-end-of-line')161 def killToEndOfLine(self, event):162 """Kill from the cursor to end of the line."""163 w = self.editWidget(event)164 if not w: return165 s = w.getAllText()166 ins = w.getInsertPoint()167 i, j = g.getLine(s, ins)168 if ins >= len(s) and g.match(s, j - 1, '\n'):169 # Kill the trailing newline of the body text.170 i = max(0, len(s) - 1)171 j = len(s)172 elif ins + 1 < j and s[ins : j - 1].strip() and g.match(s, j - 1, '\n'):173 # Kill the line, but not the newline.174 i, j = ins, j - 1175 elif g.match(s, j - 1, '\n'):176 i = ins # Kill the newline in the present line.177 else:178 i = j179 if i < j:180 self.kill(event, i, j, undoType='kill-line')181 #@+node:ekr.20150514063305.421: *3* KillLine182 @cmd('kill-line')183 def killLine(self, event):184 """Kill the line containing the cursor."""185 w = self.editWidget(event)186 if not w: return187 s = w.getAllText()188 ins = w.getInsertPoint()189 i, j = g.getLine(s, ins)190 if ins >= len(s) and g.match(s, j - 1, '\n'):191 # Kill the trailing newline of the body text.192 i = max(0, len(s) - 1)193 j = len(s)194 elif j > i + 1 and g.match(s, j - 1, '\n'):195 # Kill the line, but not the newline.196 j -= 1197 else:198 pass # Kill the newline in the present line.199 self.kill(event, i, j, undoType='kill-line')200 #@+node:ekr.20150514063305.422: *3* killRegion & killRegionSave & helper201 @cmd('kill-region')202 def killRegion(self, event):203 """Kill the text selection."""204 self.killRegionHelper(event, deleteFlag=True)205 @cmd('kill-region-save')206 def killRegionSave(self, event):207 """Add the selected text to the kill ring, but do not delete it."""208 self.killRegionHelper(event, deleteFlag=False)209 def killRegionHelper(self, event, deleteFlag):210 w = self.editWidget(event)211 if not w: return212 i, j = w.getSelectionRange()213 if i == j: return214 s = w.getSelectedText()215 if deleteFlag:216 self.beginCommand(w, undoType='kill-region')217 w.delete(i, j)218 self.endCommand(changed=True, setLabel=True)219 self.addToKillBuffer(s)220 g.app.gui.replaceClipboardWith(s)221 # self.removeRKeys(w)222 #@+node:ekr.20150514063305.423: *3* killSentence223 @cmd('kill-sentence')224 def killSentence(self, event):225 """Kill the sentence containing the cursor."""226 w = self.editWidget(event)227 if not w:228 return229 s = w.getAllText()230 ins = w.getInsertPoint()231 i = s.find('.', ins)232 if i == -1:233 return234 undoType = 'kill-sentence'235 self.beginCommand(w, undoType=undoType)236 i2 = s.rfind('.', 0, ins) + 1237 self.kill(event, i2, i + 1, undoType=undoType)238 w.setInsertPoint(i2)239 self.endCommand(changed=True, setLabel=True)240 #@+node:ekr.20150514063305.424: *3* killWs241 @cmd('kill-ws')242 def killWs(self, event, undoType='kill-ws'):243 """Kill whitespace."""244 ws = ''245 w = self.editWidget(event)246 if not w: return247 s = w.getAllText()248 i = j = ins = w.getInsertPoint()249 while i >= 0 and s[i] in (' ', '\t'):250 i -= 1251 if i < ins: i += 1...
async_process.py
Source:async_process.py
...113 """114 kill_signal = kill_signal or getattr(signal, 'SIGKILL', signal.SIGTERM)115 if self._is_running:116 LOG.debug('Halting async process [%s].', self.cmd)117 self._kill(kill_signal, kill_timeout)118 else:119 raise AsyncProcessException(_('Process is not running.'))120 if block:121 common_utils.wait_until_true(lambda: not self.is_active())122 def _spawn(self):123 """Spawn a process and its watchers."""124 self._is_running = True125 self._pid = None126 self._kill_event = eventlet.event.Event()127 self._process, cmd = utils.create_process(self._cmd,128 run_as_root=self.run_as_root)129 self._watchers = []130 for reader in (self._read_stdout, self._read_stderr):131 # Pass the stop event directly to the greenthread to132 # ensure that assignment of a new event to the instance133 # attribute does not prevent the greenthread from using134 # the original event.135 watcher = eventlet.spawn(self._watch_process,136 reader,137 self._kill_event)138 self._watchers.append(watcher)139 @property140 def pid(self):141 if self._process:142 if not self._pid:143 self._pid = utils.get_root_helper_child_pid(144 self._process.pid,145 self.cmd_without_namespace,146 run_as_root=self.run_as_root)147 return self._pid148 def _kill(self, kill_signal, kill_timeout=None):149 """Kill the process and the associated watcher greenthreads."""150 pid = self.pid151 if pid:152 self._is_running = False153 self._pid = None154 self._kill_process_and_wait(pid, kill_signal, kill_timeout)155 # Halt the greenthreads if they weren't already.156 if self._kill_event:157 self._kill_event.send()158 self._kill_event = None159 def _kill_process_and_wait(self, pid, kill_signal, kill_timeout=None):160 kill_result = self._kill_process(pid, kill_signal)161 if kill_result is False:162 return kill_result163 if self._process:164 try:165 self._process.wait(kill_timeout)166 except subprocess.TimeoutExpired:167 LOG.warning("Process %(pid)s [%(cmd)s] still running after "168 "%(timeout)d seconds. Sending %(signal)d to kill "169 "it.",170 {'pid': pid,171 'cmd': self.cmd,172 'timeout': kill_timeout,173 'signal': signal.SIGKILL})174 return self._kill_process(pid, signal.SIGKILL)175 return True176 def _kill_process(self, pid, kill_signal):177 try:178 # A process started by a root helper will be running as179 # root and need to be killed via the same helper.180 utils.kill_process(pid, kill_signal, self.run_as_root)181 except Exception:182 LOG.exception('An error occurred while killing [%s].',183 self.cmd)184 return False185 return True186 def _handle_process_error(self):187 """Kill the async process and respawn if necessary."""188 stdout = list(self.iter_stdout())189 stderr = list(self.iter_stderr())190 LOG.debug('Halting async process [%s] in response to an error. stdout:'191 ' [%s] - stderr: [%s]', self.cmd, stdout, stderr)192 self._kill(getattr(signal, 'SIGKILL', signal.SIGTERM))193 if self.respawn_interval is not None and self.respawn_interval >= 0:194 eventlet.sleep(self.respawn_interval)195 LOG.debug('Respawning async process [%s].', self.cmd)196 try:197 self.start()198 except AsyncProcessException:199 # Process was already respawned by someone else...200 pass201 def _watch_process(self, callback, kill_event):202 while not kill_event.ready():203 try:204 output = callback()205 if not output and output != "":206 break...
test_kill.py
Source:test_kill.py
1#!/usr/bin/env python32'''Kill tests.3These tests run many scenarios in which jobs are killed (KILL_JOB).4'''5import json6import pandas as pd7import pytest8from helper import *9# The number of kills to do for each job.10KillCountMode = namedtuple('KillCountMode', ['name', 'kill_count_per_job'])11# The number of seconds between a job execution and its kill.12KillDelayMode = namedtuple('KillDelayMode', ['name', 'kill_delay'])13###################################################################14# Kill a job a given amount of time after starting its execution. #15###################################################################16def kill_after_delay(platform, workload, algorithm, redis_mode, nb_kills_per_job, delay_before_kill):17 test_name = f'kill-{delay_before_kill.name}-{nb_kills_per_job.name}-{algorithm.name}-{platform.name}-{workload.name}-{redis_mode.name}'18 output_dir, robin_filename, schedconf_filename = init_instance(test_name)19 if algorithm.sched_implem != 'batsched': raise Exception('This test only supports batsched for now')20 batparams = "--forward-profiles-on-submission"21 if redis_mode.enabled == True: batparams = batparams + " --enable-redis"22 batcmd = gen_batsim_cmd(platform.filename, workload.filename, output_dir, batparams)23 schedconf_content = {24 "delay_before_kill": delay_before_kill.kill_delay,25 "nb_kills_per_job": nb_kills_per_job.kill_count_per_job,26 }27 write_file(schedconf_filename, json.dumps(schedconf_content))28 instance = RobinInstance(output_dir=output_dir,29 batcmd=batcmd,30 schedcmd=f"batsched -v '{algorithm.sched_algo_name}' --variant_options_filepath '{schedconf_filename}'",31 simulation_timeout=30, ready_timeout=5,32 success_timeout=10, failure_timeout=033 )34 instance.to_file(robin_filename)35 ret = run_robin(robin_filename)36 if ret.returncode != 0: raise Exception(f'Bad robin return code ({ret.returncode})')37 # Analyze Batsim results38 batjobs_filename = f'{output_dir}/batres_jobs.csv'39 jobs = pd.read_csv(batjobs_filename)40 epsilon = 0.141 max_runtime = delay_before_kill.kill_delay + epsilon42 jobs_too_long = jobs.loc[jobs['execution_time'] > max_runtime]43 if jobs_too_long.empty:44 print(f'All jobs have been shorter than {max_runtime} s')45 else:46 print(f'Some jobs were longer than {max_runtime} s')47 print(jobs_too_long)48 raise Exception('Some were longer than f{max_runtime} s')49@pytest.mark.parametrize("nb_kills_per_job", [KillCountMode(f'{n}kills', n) for n in [1,2]])50@pytest.mark.parametrize("delay_before_kill", [KillDelayMode(f'after{n}', n) for n in [5,10,15]])51def test_kill_after_delay(small_platform, small_workload, killer_algorithm, redis_mode, nb_kills_per_job, delay_before_kill):52 kill_after_delay(small_platform, small_workload, killer_algorithm, redis_mode, nb_kills_per_job, delay_before_kill)53@pytest.mark.parametrize("nb_kills_per_job", [KillCountMode(f'{n}kills', n) for n in [1,2]])54@pytest.mark.parametrize("delay_before_kill", [KillDelayMode(f'after{n}', n) for n in [5,10,15]])55def test_kill_after_delay_sequences(small_platform, delaysequences_workload, killer_algorithm, redis_mode, nb_kills_per_job, delay_before_kill):56 kill_after_delay(small_platform, delaysequences_workload, killer_algorithm, redis_mode, nb_kills_per_job, delay_before_kill)57@pytest.mark.parametrize("nb_kills_per_job", [KillCountMode(f'{n}kills', n) for n in [1,2]])58@pytest.mark.parametrize("delay_before_kill", [KillDelayMode(f'after{n}', n) for n in [0]])59def test_kill_after_delay0(small_platform, small_workload, killer_algorithm, redis_mode, nb_kills_per_job, delay_before_kill):60 kill_after_delay(small_platform, small_workload, killer_algorithm, redis_mode, nb_kills_per_job, delay_before_kill)61@pytest.mark.parametrize("nb_kills_per_job", [KillCountMode(f'{n}kills', n) for n in [1]])62@pytest.mark.parametrize("delay_before_kill", [KillDelayMode(f'after{n}', n) for n in [1]])63def test_kill_after_delay_smpi(small_platform, smpi_workload, killer_algorithm, redis_mode, nb_kills_per_job, delay_before_kill):64 kill_after_delay(small_platform, smpi_workload, killer_algorithm, redis_mode, nb_kills_per_job, delay_before_kill)65#####################################################66# Kill the running job when a new one is submitted. #67#####################################################68def kill_on_new_submit(platform, workload, algorithm, redis_mode):69 test_name = f'kill-onnewsubmit-{algorithm.name}-{platform.name}-{workload.name}-{redis_mode.name}'70 output_dir, robin_filename, _ = init_instance(test_name)71 if algorithm.sched_implem != 'batsched': raise Exception('This test only supports batsched for now')72 batparams = "--forward-profiles-on-submission"73 if redis_mode.enabled == True: batparams = batparams + " --enable-redis"74 batcmd = gen_batsim_cmd(platform.filename, workload.filename, output_dir, batparams)75 instance = RobinInstance(output_dir=output_dir,76 batcmd=batcmd,77 schedcmd=f"batsched -v '{algorithm.sched_algo_name}'",78 simulation_timeout=30, ready_timeout=5,79 success_timeout=10, failure_timeout=080 )81 instance.to_file(robin_filename)82 ret = run_robin(robin_filename)83 if ret.returncode != 0: raise Exception(f'Bad robin return code ({ret.returncode})')84def test_kill_on_new_submit(small_platform, small_workload, killer2_algorithm, redis_mode):85 kill_on_new_submit(small_platform, small_workload, killer2_algorithm, redis_mode)86###############################################################################87# Kill jobs after a given delay, check that the given progress is consistent. #88###############################################################################89def kill_progress(platform, workload, algorithm, redis_mode, nb_kills_per_job, delay_before_kill):90 test_name = f'kill-{delay_before_kill.name}-{nb_kills_per_job.name}-{algorithm.name}-{platform.name}-{workload.name}-{redis_mode.name}'91 output_dir, robin_filename, schedconf_filename = init_instance(test_name)92 if algorithm.sched_implem != 'batsched': raise Exception('This test only supports batsched for now')93 batparams = "--forward-profiles-on-submission"94 if redis_mode.enabled == True: batparams = batparams + " --enable-redis"95 batcmd = gen_batsim_cmd(platform.filename, workload.filename, output_dir, batparams)96 schedconf_content = {97 "delay_before_kill": delay_before_kill.kill_delay,98 "nb_kills_per_job": nb_kills_per_job.kill_count_per_job,99 }100 write_file(schedconf_filename, json.dumps(schedconf_content))101 instance = RobinInstance(output_dir=output_dir,102 batcmd=batcmd,103 schedcmd=f"batsched -v '{algorithm.sched_algo_name}' --variant_options_filepath '{schedconf_filename}'",104 simulation_timeout=30, ready_timeout=5,105 success_timeout=10, failure_timeout=0106 )107 instance.to_file(robin_filename)108 ret = run_robin(robin_filename)109 if ret.returncode != 0: raise Exception(f'Bad robin return code ({ret.returncode})')110 # Read Batsim log file.111 batlog_filename = f'{output_dir}/log/batsim.log'112 batlog_content = open(batlog_filename, 'r').read()113 # Parse messages, then events.114 messages = parse_proto_messages_from_batsim(batlog_content)115 events = retrieve_proto_events(messages)116 # Filter JOB_KILLED events.117 kill_events = [e for e in events if e['type'] == 'JOB_KILLED']118 nb_err = 0119 for e in kill_events:120 job_data = e['data']121 if 'job_progress' not in job_data:122 print('Error: a job progress is missing!', job_data)123 nb_err = nb_err + 1124 #TODO test if this job is msg or delay of composed125 #and test the progress content type in regards126 if nb_err > 0:127 raise Exception('The progress of some killed jobs was missing.')128@pytest.mark.parametrize("nb_kills_per_job", [KillCountMode(f'{n}kills', n) for n in [1]])129@pytest.mark.parametrize("delay_before_kill", [KillDelayMode(f'after{n}', n) for n in [5]])130def test_kill_progress(small_platform, one_job_workload, killer_algorithm, redis_mode, nb_kills_per_job, delay_before_kill):...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run Airtest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
