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

How to use _install_sigterm_handler method in tox

Best Python code snippet using tox_python

action.py

Source:action.py

...48        self.terminate_timeout = terminate_timeout49        if is_main_thread():50            # python allows only main thread to install signal handlers51            # see https://docs.python.org/3/library/signal.html#signals-and-threads52            self._install_sigterm_handler()53    def __enter__(self):54        msg = "{} {}".format(self.msg, " ".join(map(str, self.args)))55        self._timed_report = reporter.timed_operation(self.name, msg)56        self._timed_report.__enter__()57        return self58    def __exit__(self, type, value, traceback):59        self._timed_report.__exit__(type, value, traceback)60    def setactivity(self, name, msg):61        self.activity = name62        if msg:63            reporter.verbosity0("{} {}: {}".format(self.name, name, msg), bold=True)64        else:65            reporter.verbosity1("{} {}: {}".format(self.name, name, msg), bold=True)66    def info(self, name, msg):67        reporter.verbosity1("{} {}: {}".format(self.name, name, msg), bold=True)68    def popen(69        self,70        args,71        cwd=None,72        env=None,73        redirect=True,74        returnout=False,75        ignore_ret=False,76        capture_err=True,77        callback=None,78        report_fail=True,79    ):80        """this drives an interaction with a subprocess"""81        cwd = py.path.local() if cwd is None else cwd82        cmd_args = [str(x) for x in self._rewrite_args(cwd, args)]83        cmd_args_shell = " ".join(shlex_quote(i) for i in cmd_args)84        stream_getter = self._get_standard_streams(85            capture_err,86            cmd_args_shell,87            redirect,88            returnout,89            cwd,90        )91        exit_code, output = None, None92        with stream_getter as (fin, out_path, stderr, stdout):93            try:94                process = self.via_popen(95                    cmd_args,96                    stdout=stdout,97                    stderr=stderr,98                    cwd=str(cwd),99                    env=os.environ.copy() if env is None else env,100                    universal_newlines=True,101                    shell=False,102                    creationflags=(103                        subprocess.CREATE_NEW_PROCESS_GROUP104                        if sys.platform == "win32"105                        else 0106                        # needed for Windows signal send ability (CTRL+C)107                    ),108                )109            except OSError as exception:110                exit_code = exception.errno111            else:112                if callback is not None:113                    callback(process)114                reporter.log_popen(cwd, out_path, cmd_args_shell, process.pid)115                output = self.evaluate_cmd(fin, process, redirect)116                exit_code = process.returncode117            finally:118                if out_path is not None and out_path.exists():119                    lines = out_path.read_text("UTF-8").split("\n")120                    # first three lines are the action, cwd, and cmd - remove it121                    output = "\n".join(lines[3:])122                try:123                    if exit_code and not ignore_ret:124                        if report_fail:125                            msg = "invocation failed (exit code {:d})".format(exit_code)126                            if out_path is not None:127                                msg += ", logfile: {}".format(out_path)128                                if not out_path.exists():129                                    msg += " warning log file missing"130                            reporter.error(msg)131                            if out_path is not None and out_path.exists():132                                reporter.separator("=", "log start", Verbosity.QUIET)133                                reporter.quiet(output)134                                reporter.separator("=", "log end", Verbosity.QUIET)135                        raise InvocationError(cmd_args_shell, exit_code, output)136                finally:137                    self.command_log.add_command(cmd_args, output, exit_code)138        return output139    def evaluate_cmd(self, input_file_handler, process, redirect):140        try:141            if self.generate_tox_log and not redirect:142                if process.stderr is not None:143                    # prevent deadlock144                    raise ValueError("stderr must not be piped here")145                # we read binary from the process and must write using a binary stream146                buf = getattr(sys.stdout, "buffer", sys.stdout)147                last_time = time.time()148                while True:149                    # we have to read one byte at a time, otherwise there150                    # might be no output for a long time with slow tests151                    data = input_file_handler.read(1)152                    if data:153                        buf.write(data)154                        if b"\n" in data or (time.time() - last_time) > 1:155                            # we flush on newlines or after 1 second to156                            # provide quick enough feedback to the user157                            # when printing a dot per test158                            buf.flush()159                            last_time = time.time()160                    elif process.poll() is not None:161                        if process.stdout is not None:162                            process.stdout.close()163                        break164                    else:165                        time.sleep(0.1)166                        # the seek updates internal read buffers167                        input_file_handler.seek(0, 1)168                input_file_handler.close()169            out, _ = process.communicate()  # wait to finish170        except KeyboardInterrupt as exception:171            reporter.error("got KeyboardInterrupt signal")172            main_thread = is_main_thread()173            while True:174                try:175                    if main_thread:176                        # spin up a new thread to disable further interrupt on main thread177                        stopper = Thread(target=self.handle_interrupt, args=(process,))178                        stopper.start()179                        stopper.join()180                    else:181                        self.handle_interrupt(process)182                except KeyboardInterrupt:183                    continue184                break185            raise exception186        return out187    def handle_interrupt(self, process):188        """A three level stop mechanism for children - INT -> TERM -> KILL"""189        msg = "from {} {{}} pid {}".format(os.getpid(), process.pid)190        if self._wait(process, self.suicide_timeout) is None:191            self.info("KeyboardInterrupt", msg.format("SIGINT"))192            process.send_signal(signal.CTRL_C_EVENT if sys.platform == "win32" else signal.SIGINT)193            if self._wait(process, self.interrupt_timeout) is None:194                self.info("KeyboardInterrupt", msg.format("SIGTERM"))195                process.terminate()196                if self._wait(process, self.terminate_timeout) is None:197                    self.info("KeyboardInterrupt", msg.format("SIGKILL"))198                    process.kill()199                    process.communicate()200    @staticmethod201    def _wait(process, timeout):202        if sys.version_info >= (3, 3):203            # python 3 has timeout feature built-in204            try:205                process.communicate(timeout=timeout)206            except subprocess.TimeoutExpired:207                pass208        else:209            # on Python 2 we need to simulate it210            delay = 0.01211            while process.poll() is None and timeout > 0:212                time.sleep(delay)213                timeout -= delay214        return process.poll()215    @contextmanager216    def _get_standard_streams(self, capture_err, cmd_args_shell, redirect, returnout, cwd):217        stdout = out_path = input_file_handler = None218        stderr = subprocess.STDOUT if capture_err else None219        if self.generate_tox_log or redirect:220            out_path = self.get_log_path(self.name)221            with out_path.open("wt") as stdout, out_path.open("rb") as input_file_handler:222                msg = "action: {}, msg: {}\ncwd: {}\ncmd: {}\n".format(223                    self.name.replace("\n", " "),224                    self.msg.replace("\n", " "),225                    str(cwd).replace("\n", " "),226                    cmd_args_shell.replace("\n", " "),227                )228                stdout.write(msg)229                stdout.flush()230                input_file_handler.read()  # read the header, so it won't be written to stdout231                yield input_file_handler, out_path, stderr, stdout232                return233        if returnout:234            stdout = subprocess.PIPE235        yield input_file_handler, out_path, stderr, stdout236    def get_log_path(self, actionid):237        log_file = get_unique_file(self.log_dir, prefix=actionid, suffix=".log")238        return log_file239    def _rewrite_args(self, cwd, args):240        executable = None241        if INFO.IS_WIN:242            # shebang lines are not adhered on Windows so if it's a python script243            # pre-pend the interpreter244            ext = os.path.splitext(str(args[0]))[1].lower()245            if ext == ".py":246                executable = str(self.python)247        if executable is None:248            executable = args[0]249            args = args[1:]250        new_args = [executable]251        # to make the command shorter try to use relative paths for all subsequent arguments252        # note the executable cannot be relative as the Windows applies cwd after invocation253        for arg in args:254            if arg and os.path.isabs(str(arg)):255                arg_path = py.path.local(arg)256                if arg_path.exists() and arg_path.common(cwd) is not None:257                    potential_arg = cwd.bestrelpath(arg_path)258                    if len(potential_arg.split("..")) < 2:259                        # just one parent directory accepted as relative path260                        arg = potential_arg261            new_args.append(str(arg))262        return new_args263    def _install_sigterm_handler(self):264        """Handle sigterm as if it were a keyboardinterrupt"""265        def sigterm_handler(signum, frame):266            reporter.error("Got SIGTERM, handling it as a KeyboardInterrupt")267            raise KeyboardInterrupt()...

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.