clean up env.run(..., call=True)

- have it return a string, simplifying the type annotations
- have it check for error

src/poetry/utils/env.py

@@ -1499,16 +1499,16 @@ def get_command_from_bin(self, bin: str) -> list[str]:
 
  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 @@ def run_python_script(self, content: str, **kwargs: Any) -> int | str:
  **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 @@ def _run(self, cmd: list[str], **kwargs: Any) -> int | str:
  ).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:
@@ -1723,19 +1724,16 @@ def __init__(self, path: Path, base: Path | None = None) -> None:
  # from inside the virtualenv.
  if base is None:
  output = self.run_python_script(GET_BASE_PREFIX)
- assert isinstance(output, str)
  self._base = Path(output.strip())
 
  @property
  def sys_path(self) -> list[str]:
  output = self.run_python_script(GET_SYS_PATH)
- assert isinstance(output, str)
  paths: list[str] = json.loads(output)
  return paths
 
  def get_version_info(self) -> tuple[Any, ...]:
  output = self.run_python_script(GET_PYTHON_VERSION)
- assert isinstance(output, str)
 
  return tuple(int(s) for s in output.strip().split("."))
 
@@ -1745,20 +1743,17 @@ def get_python_implementation(self) -> str:
 
  def get_supported_tags(self) -> list[Tag]:
  output = self.run_python_script(GET_SYS_TAGS)
- assert isinstance(output, str)
 
  return [Tag(*t) for t in json.loads(output)]
 
  def get_marker_env(self) -> dict[str, Any]:
  output = self.run_python_script(GET_ENVIRONMENT_INFO)
- assert isinstance(output, str)
 
  env: dict[str, Any] = json.loads(output)
  return env
 
  def get_pip_version(self) -> Version:
  output = self.run_pip("--version")
- assert isinstance(output, str)
  output = output.strip()
 
  m = re.match("pip (.+?)(?: from .+)?$", output)
@@ -1769,7 +1764,6 @@ def get_pip_version(self) -> Version:
 
  def get_paths(self) -> dict[str, str]:
  output = self.run_python_script(GET_PATHS)
- assert isinstance(output, str)
  paths: dict[str, str] = json.loads(output)
  return paths
 
@@ -1780,7 +1774,7 @@ def is_sane(self) -> bool:
  # 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)
 
@@ -1885,7 +1879,6 @@ def find_executables(self) -> None:
 
  def get_paths(self) -> dict[str, str]:
  output = self.run_python_script(GET_PATHS_FOR_GENERIC_ENVS)
- assert isinstance(output, str)
 
  paths: dict[str, str] = json.loads(output)
  return paths
@@ -1902,7 +1895,7 @@ def execute(self, bin: str, *args: str, **kwargs: Any) -> int:
 
  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 +1925,12 @@ def paths(self) -> dict[str, str]:
 
  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))

src/poetry/utils/pip.py

@@ -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

tests/puzzle/test_provider.py

@@ -36,7 +36,7 @@
 
 
 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=""))
 
 

tests/utils/test_env.py

@@ -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,15 +1010,15 @@ 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"
  ),
  )
  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 @@ def test_call_does_not_block_on_full_pipe(
  )
 
  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()