How to use sigterm_handler method in tox

Best Python code snippet using tox_python

base_worker.py

Source:base_worker.py Github

copy

Full Screen

...32 # NOTE: There are several cases when the handler will not have33 # up-to-date information on running the command and/or process,34 # but chances of that happening are VERY slim and the35 # consequences are not fatal.36 def sigterm_handler(signal, frame):37 if sigterm_handler.triggered:38 return39 else:40 sigterm_handler.triggered = True41 if sigterm_handler.proc is not None:42 sigterm_handler.proc.wait()43 if sigterm_handler.command is not None:44 command_manager.set_running_command_as_pending(sigterm_handler.command)45 sys.exit(0)46 sigterm_handler.triggered = False47 sigterm_handler.command = None48 sigterm_handler.proc = None49 signal.signal(signal.SIGTERM, sigterm_handler)50 while True:...

Full Screen

Full Screen

signals.py

Source:signals.py Github

copy

Full Screen

1import asyncio2import logging3import signal4import threading5from contextlib import contextmanager6from typing import Any, Callable, Optional7logger = logging.getLogger(__name__)8# A few things about signals and asyncio:9# - https://docs.python.org/3/library/asyncio-eventloop.html#unix-signals10# - If you don't use loop.add_signal_handler, your handler has no effect on the queue11# - asyncio signal primitives don't give you a way to get the current handler for a12# signal13# - If a signal has an asyncio handler, its "standard handler (installed through14# signal.signal) is ignored (you can't get it, you can't set one, it doesn't get15# called on the signal).16# - asyncio handlers receive no parameter17# This all mean we have to save the previous signals before installing ours, and for18# uninstalling, we need to remove the async handler and only then re-add the normal19# one. And hope that there was no previously set async handler.20@contextmanager21def on_stop(callback: Callable[[], None]):22 if threading.current_thread() is not threading.main_thread():23 logger.warning(24 "Skipping signal handling, because this is not the main thread",25 extra={"action": "skip_signal_handlers"},26 )27 yield28 return29 sigint_handler = signal.getsignal(signal.SIGINT)30 sigterm_handler = signal.getsignal(signal.SIGTERM)31 uninstalled = False32 loop: Optional[asyncio.AbstractEventLoop]33 try:34 loop = asyncio.get_running_loop()35 except RuntimeError:36 loop = None37 def uninstall_and_callback(*args) -> None:38 nonlocal uninstalled39 uninstalled = True40 uninstall(41 loop=loop, sigint_handler=sigint_handler, sigterm_handler=sigterm_handler42 )43 callback()44 try:45 install(loop=loop, handler=uninstall_and_callback)46 yield47 finally:48 if not uninstalled:49 uninstall(50 loop=loop,51 sigint_handler=sigint_handler,52 sigterm_handler=sigterm_handler,53 )54def install(loop: Optional[asyncio.AbstractEventLoop], handler: Callable):55 logger.debug(56 "Installing signal handler", extra={"action": "install_signal_handlers"}57 )58 if loop:59 loop.add_signal_handler(signal.SIGINT, handler)60 loop.add_signal_handler(signal.SIGTERM, handler)61 else:62 signal.signal(signal.SIGINT, handler)63 signal.signal(signal.SIGTERM, handler)64def uninstall(65 loop: Optional[asyncio.AbstractEventLoop],66 sigint_handler: Any,67 sigterm_handler: Any,68):69 logger.debug(70 "Resetting previous signal handler",71 extra={"action": "uninstall_signal_handlers"},72 )73 if loop:74 loop.remove_signal_handler(signal.SIGINT)75 loop.remove_signal_handler(signal.SIGTERM)76 signal.signal(signal.SIGINT, sigint_handler)...

Full Screen

Full Screen

ray_process_reaper.py

Source:ray_process_reaper.py Github

copy

Full Screen

...15to SIGKILL).16"""17SIGTERM_GRACE_PERIOD_SECONDS = 118def reap_process_group(*args):19 def sigterm_handler(*args):20 # Give a one-second grace period for other processes to clean up.21 time.sleep(SIGTERM_GRACE_PERIOD_SECONDS)22 # SIGKILL the pgroup (including ourselves) as a last-resort.23 if sys.platform == "win32":24 atexit.unregister(sigterm_handler)25 os.kill(0, signal.CTRL_BREAK_EVENT)26 else:27 os.killpg(0, signal.SIGKILL)28 # Set a SIGTERM handler to handle SIGTERMing ourselves with the group.29 if sys.platform == "win32":30 atexit.register(sigterm_handler)31 else:32 signal.signal(signal.SIGTERM, sigterm_handler)33 # Our parent must have died, SIGTERM the group (including ourselves)....

Full Screen

Full Screen

process.py

Source:process.py Github

copy

Full Screen

1from time import sleep2import sys3import signal4def sigterm_handler(_signo, _stack_frame):5 print("sigterm_handler executed, %s, %s" % (_signo, _stack_frame))6 sys.exit(0)7if __name__ == "__main__":8 signal.signal(signal.SIGTERM, sigterm_handler)9 try:10 while True:11 print("Doing something")12 sleep(3)13 finally:...

Full Screen

Full Screen

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:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run tox 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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful