diff --git a/src/poetry/utils/env.py b/src/poetry/utils/env.py index 47d5ac62a3f..6510f897db4 100644 --- a/src/poetry/utils/env.py +++ b/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)) diff --git a/src/poetry/utils/pip.py b/src/poetry/utils/pip.py index eb38bbf0b3b..b74294795ed 100644 --- a/src/poetry/utils/pip.py +++ b/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 diff --git a/tests/puzzle/test_provider.py b/tests/puzzle/test_provider.py index 1264321b9d0..dd374a23743 100644 --- a/tests/puzzle/test_provider.py +++ b/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="")) diff --git a/tests/utils/test_env.py b/tests/utils/test_env.py index 6cc3deb79bd..4e1acb4b98c 100644 --- a/tests/utils/test_env.py +++ b/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,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 @@ 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()