Best Python code snippet using avocado_python
task_6.py
Source:task_6.py
...24 def is_empty(self):25 return True if (self.base_queue == [] and self.correction_queue == []) else False26 def to_queue(self, obj):27 self.base_queue.append(obj)28 def finish_task(self, from_base=True, from_correction=False):29 if from_correction:30 from_base = False31 task_to_finish = self.correction_queue.pop(0)32 if from_base:33 task_to_finish = self.base_queue.pop(0)34 self.finished_tasks.append(task_to_finish)35 def to_correction_queue(self):36 task_to_correction = self.base_queue.pop(0)37 self.correction_queue.append(task_to_correction)38 def clear_finished(self):39 self.finished_tasks.clear()40if __name__ == '__main__':41 task_queue = TaskQueue()42 task_queue.to_queue(1)43 task_queue.to_queue(2)44 task_queue.to_queue(3)45 print(f'is task queue empty?: {task_queue.is_empty()}')46 print(task_queue)47 print()48 task_queue.to_queue(4)49 task_queue.to_queue(5)50 task_queue.finish_task()51 task_queue.finish_task()52 task_queue.finish_task()53 task_queue.to_queue(6)54 task_queue.to_queue(7)55 task_queue.to_correction_queue()56 task_queue.to_correction_queue()57 task_queue.finish_task(from_correction=True)58 task_queue.finish_task(from_correction=True)59 task_queue.finish_task()60 task_queue.finish_task()61 print(f'is task queue empty?: {task_queue.is_empty()}')62 print(task_queue)63 print()64 task_queue.clear_finished()65 print(task_queue)...
slave.py
Source:slave.py
...55 timer.setName(taskname)56 timer.start();57#called when timer ends58@app.route('/finish')59def finish_task():60 global task_status 61 task_status = "finished"62 ret = stat()63 try:64 res = requests.get('http://master:8001/finish_task', data= ret)65 except:66 print("master unreached")67 count = 268 while count > 0:69 time.sleep(1)70 count -= 171 finish_task()72 return73 if(res.text) == "task_complete":74 return '0'75 else:76 print("master failed to update")77 count = 278 while count > 0:79 time.sleep(1)80 count -= 181 finish_task()82 return83def status(task_name,task_stat):84 returnJson = json.dumps({ "taskname":task_name,"status":task_stat,"host":slavename})85 return returnJson86 87if __name__ == "__main__":...
test_graceful_shutdown.py
Source:test_graceful_shutdown.py
1import asyncio2import os3import signal4from unittest import IsolatedAsyncioTestCase, mock5from ustack_tornado_shutdown import graceful_shutdown6class TestGracefulShutdown(IsolatedAsyncioTestCase):7 async def send_sigterm(self, wait_task: asyncio.Task[None]) -> bool:8 done = wait_task.done()9 os.kill(os.getpid(), signal.SIGTERM)10 return done11 @mock.patch("asyncio.sleep")12 async def test_on_sigterm_returns_after_sigterm_is_trapped(13 self,14 mock_sleep: mock.Mock15 ) -> None:16 wait_task = asyncio.create_task(graceful_shutdown.on_sigterm())17 finish_task = asyncio.create_task(self.send_sigterm(wait_task))18 await wait_task19 assert not await finish_task20 mock_sleep.assert_not_called()21 @mock.patch("asyncio.sleep")22 async def test_on_sigterm_calls_function_before_returning_when_specified(23 self,24 mock_sleep: mock.Mock25 ) -> None:26 fn = mock.Mock()27 wait_task = asyncio.create_task(graceful_shutdown.on_sigterm(fn))28 finish_task = asyncio.create_task(self.send_sigterm(wait_task))29 await wait_task30 await finish_task31 fn.assert_called_once_with()...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!