Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use sigterm_handler method in tox

Best Python code snippet using tox_python

base_worker.py

Source:base_worker.py

...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:...

signals.py

Source:signals.py

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)...

ray_process_reaper.py

Source:ray_process_reaper.py

...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)....

process.py

Source:process.py

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:...

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.