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

How to use _rewrite_args method in tox

Best Python code snippet using tox_python

action.py

Source:action.py

...57        report_fail=True,58    ):59        """this drives an interaction with a subprocess"""60        cwd = py.path.local() if cwd is None else cwd61        cmd_args = [str(x) for x in self._rewrite_args(cwd, args)]62        cmd_args_shell = " ".join(pipes.quote(i) for i in cmd_args)63        stream_getter = self._get_standard_streams(64            capture_err, cmd_args_shell, redirect, returnout, cwd65        )66        exit_code, output = None, None67        with stream_getter as (fin, out_path, stderr, stdout):68            try:69                process = self.via_popen(70                    cmd_args,71                    stdout=stdout,72                    stderr=stderr,73                    cwd=str(cwd),74                    env=os.environ.copy() if env is None else env,75                    universal_newlines=True,76                    shell=False,77                    creationflags=(78                        subprocess.CREATE_NEW_PROCESS_GROUP79                        if sys.platform == "win32"80                        else 081                        # needed for Windows signal send ability (CTRL+C)82                    ),83                )84            except OSError as exception:85                exit_code = exception.errno86            else:87                if callback is not None:88                    callback(process)89                reporter.log_popen(cwd, out_path, cmd_args_shell, process.pid)90                output = self.evaluate_cmd(fin, process, redirect)91                exit_code = process.returncode92            finally:93                if out_path is not None and out_path.exists():94                    lines = out_path.read_text("UTF-8").split("\n")95                    # first three lines are the action, cwd, and cmd - remove it96                    output = "\n".join(lines[3:])97                try:98                    if exit_code and not ignore_ret:99                        if report_fail:100                            msg = "invocation failed (exit code {:d})".format(exit_code)101                            if out_path is not None:102                                msg += ", logfile: {}".format(out_path)103                                if not out_path.exists():104                                    msg += " warning log file missing"105                            reporter.error(msg)106                            if out_path is not None and out_path.exists():107                                reporter.separator("=", "log start", Verbosity.QUIET)108                                reporter.quiet(output)109                                reporter.separator("=", "log end", Verbosity.QUIET)110                        raise InvocationError(cmd_args_shell, exit_code, output)111                finally:112                    self.command_log.add_command(cmd_args, output, exit_code)113        return output114    def evaluate_cmd(self, input_file_handler, process, redirect):115        try:116            if self.generate_tox_log and not redirect:117                if process.stderr is not None:118                    # prevent deadlock119                    raise ValueError("stderr must not be piped here")120                # we read binary from the process and must write using a binary stream121                buf = getattr(sys.stdout, "buffer", sys.stdout)122                last_time = time.time()123                while True:124                    # we have to read one byte at a time, otherwise there125                    # might be no output for a long time with slow tests126                    data = input_file_handler.read(1)127                    if data:128                        buf.write(data)129                        if b"\n" in data or (time.time() - last_time) > 1:130                            # we flush on newlines or after 1 second to131                            # provide quick enough feedback to the user132                            # when printing a dot per test133                            buf.flush()134                            last_time = time.time()135                    elif process.poll() is not None:136                        if process.stdout is not None:137                            process.stdout.close()138                        break139                    else:140                        time.sleep(0.1)141                        # the seek updates internal read buffers142                        input_file_handler.seek(0, 1)143                input_file_handler.close()144            out, _ = process.communicate()  # wait to finish145        except KeyboardInterrupt as exception:146            reporter.error("got KeyboardInterrupt signal")147            main_thread = is_main_thread()148            while True:149                try:150                    if main_thread:151                        # spin up a new thread to disable further interrupt on main thread152                        stopper = Thread(target=self.handle_interrupt, args=(process,))153                        stopper.start()154                        stopper.join()155                    else:156                        self.handle_interrupt(process)157                except KeyboardInterrupt:158                    continue159                break160            raise exception161        return out162    def handle_interrupt(self, process):163        """A three level stop mechanism for children - INT -> TERM -> KILL"""164        msg = "from {} {{}} pid {}".format(os.getpid(), process.pid)165        if process.poll() is None:166            self.info("KeyboardInterrupt", msg.format("SIGINT"))167            process.send_signal(signal.CTRL_C_EVENT if sys.platform == "win32" else signal.SIGINT)168            if self._wait(process, WAIT_INTERRUPT) is None:169                self.info("KeyboardInterrupt", msg.format("SIGTERM"))170                process.terminate()171                if self._wait(process, WAIT_TERMINATE) is None:172                    self.info("KeyboardInterrupt", msg.format("SIGKILL"))173                    process.kill()174                    process.communicate()175    @staticmethod176    def _wait(process, timeout):177        if sys.version_info >= (3, 3):178            # python 3 has timeout feature built-in179            try:180                process.communicate(timeout=WAIT_INTERRUPT)181            except subprocess.TimeoutExpired:182                pass183        else:184            # on Python 2 we need to simulate it185            delay = 0.01186            while process.poll() is None and timeout > 0:187                time.sleep(delay)188                timeout -= delay189        return process.poll()190    @contextmanager191    def _get_standard_streams(self, capture_err, cmd_args_shell, redirect, returnout, cwd):192        stdout = out_path = input_file_handler = None193        stderr = subprocess.STDOUT if capture_err else None194        if self.generate_tox_log or redirect:195            out_path = self.get_log_path(self.name)196            with out_path.open("wt") as stdout, out_path.open("rb") as input_file_handler:197                msg = "action: {}, msg: {}\ncwd: {}\ncmd: {}\n".format(198                    self.name.replace("\n", " "),199                    self.msg.replace("\n", " "),200                    str(cwd).replace("\n", " "),201                    cmd_args_shell.replace("\n", " "),202                )203                stdout.write(msg)204                stdout.flush()205                input_file_handler.read()  # read the header, so it won't be written to stdout206                yield input_file_handler, out_path, stderr, stdout207                return208        if returnout:209            stdout = subprocess.PIPE210        yield input_file_handler, out_path, stderr, stdout211    def get_log_path(self, actionid):212        log_file = get_unique_file(self.log_dir, prefix=actionid, suffix=".log")213        return log_file214    def _rewrite_args(self, cwd, args):215        executable = None216        if INFO.IS_WIN:217            # shebang lines are not adhered on Windows so if it's a python script218            # pre-pend the interpreter219            ext = os.path.splitext(str(args[0]))[1].lower()220            if ext == ".py":221                executable = str(self.python)222        if executable is None:223            executable = args[0]224            args = args[1:]225        new_args = [executable]226        # to make the command shorter try to use relative paths for all subsequent arguments227        # note the executable cannot be relative as the Windows applies cwd after invocation228        for arg in args:...

sapling_shell.py

Source:sapling_shell.py

...47                raise ValueError(f"expected more args: {args}")48            args[1] = self._get_origin()49        elif match_args(["reset"], args):50            raise ValueError(f"unexpected use of `git reset` in SaplingShell: {' '.join(args)}")51        git_args = self._rewrite_args(args)52        full_args = ["--git-dir", self.git_dir] + git_args53        return super().git(*full_args, **kwargs)54    def _rewrite_args(self, _args: List[str]) -> List[str]:55        args = _args[:]56        # When running queries against a bare repo via `git --git-dir`, Git will57        # not be able to resolve arguments like HEAD, so we must resolve those58        # to a full hash before running Git.59        if 'HEAD' in args:60            top = self._run_sapling_command(['log', '-r', 'max(descendants(.))', '-T', '{node}'])61            for index, arg in enumerate(args):62                if arg == 'HEAD':63                    args[index] = top64        return args65    def _get_origin(self):66        # This should be good enough, right???67        return self._run_sapling_command(["config", "paths.default"])68    def run_sapling_command(self, *args: str) -> str:...

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.