(
cmd: Iterable[str], installer_input=None, check=True, timeout=420, **env_vars
)
| 112 | |
| 113 | |
| 114 | def _execute( |
| 115 | cmd: Iterable[str], installer_input=None, check=True, timeout=420, **env_vars |
| 116 | ) -> subprocess.CompletedProcess: |
| 117 | t0 = time.time() |
| 118 | # The environment is not copied on Windows, so copy here to get consistent behavior |
| 119 | env = os.environ.copy() |
| 120 | if env_vars: |
| 121 | env.update({k: v for (k, v) in env_vars.items() if v is not None}) |
| 122 | p = subprocess.Popen( |
| 123 | cmd, |
| 124 | stdin=subprocess.PIPE if installer_input else None, |
| 125 | stdout=subprocess.PIPE, |
| 126 | stderr=subprocess.PIPE, |
| 127 | text=True, |
| 128 | env=env, |
| 129 | ) |
| 130 | stdout, stderr = None, None |
| 131 | try: |
| 132 | stdout, stderr = p.communicate(input=installer_input, timeout=timeout) |
| 133 | retcode = p.poll() |
| 134 | if check and retcode: |
| 135 | raise subprocess.CalledProcessError(retcode, cmd, output=stdout, stderr=stderr) |
| 136 | return subprocess.CompletedProcess(cmd, retcode, stdout, stderr) |
| 137 | except subprocess.TimeoutExpired: |
| 138 | p.kill() |
| 139 | stdout, stderr = p.communicate() |
| 140 | raise |
| 141 | finally: |
| 142 | if stdout: |
| 143 | print(stdout) |
| 144 | if stderr: |
| 145 | print(stderr, file=sys.stderr) |
| 146 | print("Took", timedelta(seconds=time.time() - t0)) |
| 147 | |
| 148 | |
| 149 | def _check_installer_log(install_dir): |
no outgoing calls
no test coverage detected