How to use _rewrite_args method in tox

Best Python code snippet using tox_python

action.py

Source:action.py Github

copy

Full Screen

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

Full Screen

Full Screen

sapling_shell.py

Source:sapling_shell.py Github

copy

Full Screen

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

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