Commit 6eccf926 by David Hotham Committed by GitHub

clarify use of subprocess.call() (#7812)

parent bc6e84e3
......@@ -1499,16 +1499,16 @@ class Env:
return [str(self._bin(bin))]
def run(self, bin: str, *args: str, **kwargs: Any) -> str | int:
def run(self, bin: str, *args: str, **kwargs: Any) -> str:
cmd = self.get_command_from_bin(bin) + list(args)
return self._run(cmd, **kwargs)
def run_pip(self, *args: str, **kwargs: Any) -> int | str:
def run_pip(self, *args: str, **kwargs: Any) -> str:
pip = self.get_pip_command()
cmd = pip + list(args)
return self._run(cmd, **kwargs)
def run_python_script(self, content: str, **kwargs: Any) -> int | str:
def run_python_script(self, content: str, **kwargs: Any) -> str:
return self.run(
self._executable,
"-I",
......@@ -1520,7 +1520,7 @@ class Env:
**kwargs,
)
def _run(self, cmd: list[str], **kwargs: Any) -> int | str:
def _run(self, cmd: list[str], **kwargs: Any) -> str:
"""
Run a command inside the Python environment.
"""
......@@ -1542,7 +1542,8 @@ class Env:
).stdout
elif call:
assert stderr != subprocess.PIPE
return subprocess.call(cmd, stderr=stderr, env=env, **kwargs)
subprocess.check_call(cmd, stderr=stderr, env=env, **kwargs)
output = ""
else:
output = subprocess.check_output(cmd, stderr=stderr, env=env, **kwargs)
except CalledProcessError as e:
......@@ -1780,7 +1781,7 @@ class VirtualEnv(Env):
# A virtualenv is considered sane if "python" exists.
return os.path.exists(self.python)
def _run(self, cmd: list[str], **kwargs: Any) -> int | str:
def _run(self, cmd: list[str], **kwargs: Any) -> str:
kwargs["env"] = self.get_temp_environ(environ=kwargs.get("env"))
return super()._run(cmd, **kwargs)
......@@ -1902,7 +1903,7 @@ class GenericEnv(VirtualEnv):
return exe.returncode
def _run(self, cmd: list[str], **kwargs: Any) -> int | str:
def _run(self, cmd: list[str], **kwargs: Any) -> str:
return super(VirtualEnv, self)._run(cmd, **kwargs)
def is_venv(self) -> bool:
......@@ -1932,12 +1933,12 @@ class NullEnv(SystemEnv):
return self._paths
def _run(self, cmd: list[str], **kwargs: Any) -> int | str:
def _run(self, cmd: list[str], **kwargs: Any) -> str:
self.executed.append(cmd)
if self._execute:
return super()._run(cmd, **kwargs)
return 0
return ""
def execute(self, bin: str, *args: str, **kwargs: Any) -> int:
self.executed.append([bin] + list(args))
......
......@@ -18,7 +18,7 @@ def pip_install(
editable: bool = False,
deps: bool = False,
upgrade: bool = False,
) -> int | str:
) -> str:
is_wheel = path.suffix == ".whl"
# We disable version check here as we are already pinning to version available in
......
......@@ -36,7 +36,7 @@ SOME_URL = "https://example.com/path.tar.gz"
class MockEnv(BaseMockEnv):
def run(self, bin: str, *args: str, **kwargs: Any) -> str | int:
def run(self, bin: str, *args: str, **kwargs: Any) -> str:
raise EnvCommandError(CalledProcessError(1, "python", output=""))
......
......@@ -966,11 +966,11 @@ def test_call_with_input_and_keyboard_interrupt(
def test_call_no_input_with_keyboard_interrupt(
tmp_path: Path, tmp_venv: VirtualEnv, mocker: MockerFixture
) -> None:
mocker.patch("subprocess.call", side_effect=KeyboardInterrupt())
mocker.patch("subprocess.check_call", side_effect=KeyboardInterrupt())
kwargs = {"call": True}
with pytest.raises(KeyboardInterrupt):
tmp_venv.run("python", "-", **kwargs)
subprocess.call.assert_called_once()
subprocess.check_call.assert_called_once()
def test_run_with_called_process_error(
......@@ -1010,7 +1010,7 @@ def test_call_no_input_with_called_process_error(
tmp_path: Path, tmp_venv: VirtualEnv, mocker: MockerFixture
) -> None:
mocker.patch(
"subprocess.call",
"subprocess.check_call",
side_effect=subprocess.CalledProcessError(
42, "some_command", "some output", "some error"
),
......@@ -1018,7 +1018,7 @@ def test_call_no_input_with_called_process_error(
kwargs = {"call": True}
with pytest.raises(EnvCommandError) as error:
tmp_venv.run("python", "-", **kwargs)
subprocess.call.assert_called_once()
subprocess.check_call.assert_called_once()
assert "some output" in str(error.value)
assert "some error" in str(error.value)
......@@ -1054,9 +1054,10 @@ for i in range(10000):
)
def target(result: list[int]) -> None:
result.append(tmp_venv.run("python", str(script), call=True))
tmp_venv.run("python", str(script), call=True)
result.append(0)
results = []
results: list[int] = []
# use a separate thread, so that the test does not block in case of error
thread = Thread(target=target, args=(results,))
thread.start()
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment