Tests for ansi and no-ansi installs + bugfix for ansi install (#3881)

* add tests for ansi and no-ansi install output

* add tests for Env.run error handling

* add tests for pip_install error handling

* encapsulate pip installation to handle interrupts

poetry/installation/executor.py

@@ -117,6 +117,26 @@ class Executor:
 
         return self
 
+    def pip_install(
+        self, req: Union[Path, str], upgrade: bool = False, editable: bool = False
+    ) -> int:
+        func = pip_install
+        if editable:
+            func = pip_editable_install
+
+        try:
+            func(req, self._env, upgrade=upgrade)
+        except EnvCommandError as e:
+            output = decode(e.e.output)
+            if (
+                "KeyboardInterrupt" in output
+                or "ERROR: Operation cancelled by user" in output
+            ):
+                return -2
+            raise
+
+        return 0
+
     def execute(self, operations: List["OperationTypes"]) -> int:
         self._total_operations = len(operations)
         for job_type in self._executed:
@@ -483,9 +503,7 @@ class Executor:
             )
         )
         self._write(operation, message)
-        return pip_install(
-            str(archive), self._env, upgrade=operation.job_type == "update"
-        )
+        return self.pip_install(str(archive), upgrade=operation.job_type == "update")
 
     def _update(self, operation: Union[Install, Update]) -> int:
         return self._install(operation)
@@ -577,13 +595,13 @@ class Executor:
 
                 with builder.setup_py():
                     if package.develop:
-                        return pip_editable_install(req, self._env)
-                    return pip_install(req, self._env, upgrade=True)
+                        return self.pip_install(req, editable=True)
+                    return self.pip_install(req, upgrade=True)
 
         if package.develop:
-            return pip_editable_install(req, self._env)
+            return self.pip_install(req, editable=True)
 
-        return pip_install(req, self._env, upgrade=True)
+        return self.pip_install(req, upgrade=True)
 
     def _install_git(self, operation: Union[Install, Update]) -> int:
         from poetry.core.vcs import Git

tests/installation/test_executor.py

@@ -44,6 +44,22 @@ def io():
 
 
 @pytest.fixture()
+def io_decorated():
+    io = BufferedIO(decorated=True)
+    io.output.formatter.set_style("c1", Style("cyan"))
+    io.output.formatter.set_style("success", Style("green"))
+
+    return io
+
+
+@pytest.fixture()
+def io_not_decorated():
+    io = BufferedIO(decorated=False)
+
+    return io
+
+
+@pytest.fixture()
 def pool():
     pool = Pool()
     pool.add_repository(MockRepository())
@@ -187,6 +203,70 @@ Package operations: 1 install, 0 updates, 0 removals
     assert expected in io.fetch_output()
 
 
+def test_execute_works_with_ansi_output(
+    mocker, config, pool, io_decorated, tmp_dir, mock_file_downloads, env
+):
+    config = Config()
+    config.merge({"cache-dir": tmp_dir})
+
+    executor = Executor(env, pool, config, io_decorated)
+
+    install_output = (
+        "some string that does not contain a keyb0ard !nterrupt or cance11ed by u$er"
+    )
+    mocker.patch.object(env, "_run", return_value=install_output)
+    return_code = executor.execute(
+        [
+            Install(Package("pytest", "3.5.2")),
+        ]
+    )
+    env._run.assert_called_once()
+
+    expected = [
+        "\x1b[39;1mPackage operations\x1b[39;22m: \x1b[34m1\x1b[39m install, \x1b[34m0\x1b[39m updates, \x1b[34m0\x1b[39m removals",
+        "\x1b[34;1m•\x1b[39;22m \x1b[39mInstalling \x1b[39m\x1b[36mpytest\x1b[39m\x1b[39m (\x1b[39m\x1b[39;1m3.5.2\x1b[39;22m\x1b[39m)\x1b[39m: \x1b[34mPending...\x1b[39m",
+        "\x1b[34;1m•\x1b[39;22m \x1b[39mInstalling \x1b[39m\x1b[36mpytest\x1b[39m\x1b[39m (\x1b[39m\x1b[39;1m3.5.2\x1b[39;22m\x1b[39m)\x1b[39m: \x1b[34mDownloading...\x1b[39m",
+        "\x1b[34;1m•\x1b[39;22m \x1b[39mInstalling \x1b[39m\x1b[36mpytest\x1b[39m\x1b[39m (\x1b[39m\x1b[39;1m3.5.2\x1b[39;22m\x1b[39m)\x1b[39m: \x1b[34mInstalling...\x1b[39m",
+        "\x1b[32;1m•\x1b[39;22m \x1b[39mInstalling \x1b[39m\x1b[36mpytest\x1b[39m\x1b[39m (\x1b[39m\x1b[32m3.5.2\x1b[39m\x1b[39m)\x1b[39m",  # finished
+    ]
+    output = io_decorated.fetch_output()
+    # hint: use print(repr(output)) if you need to debug this
+
+    for line in expected:
+        assert line in output
+    assert 0 == return_code
+
+
+def test_execute_works_with_no_ansi_output(
+    mocker, config, pool, io_not_decorated, tmp_dir, mock_file_downloads, env
+):
+    config = Config()
+    config.merge({"cache-dir": tmp_dir})
+
+    executor = Executor(env, pool, config, io_not_decorated)
+
+    install_output = (
+        "some string that does not contain a keyb0ard !nterrupt or cance11ed by u$er"
+    )
+    mocker.patch.object(env, "_run", return_value=install_output)
+    return_code = executor.execute(
+        [
+            Install(Package("pytest", "3.5.2")),
+        ]
+    )
+    env._run.assert_called_once()
+
+    expected = """
+Package operations: 1 install, 0 updates, 0 removals
+
+  • Installing pytest (3.5.2)
+"""
+    expected = set(expected.splitlines())
+    output = set(io_not_decorated.fetch_output().splitlines())
+    assert expected == output
+    assert 0 == return_code
+
+
 def test_execute_should_show_operation_as_cancelled_on_subprocess_keyboard_interrupt(
     config, mocker, io, env
 ):

tests/utils/test_env.py

 import os
 import shutil
+import subprocess
 import sys
 
 from pathlib import Path
@@ -638,6 +639,58 @@ def test_run_with_input_non_zero_return(tmp_dir, tmp_venv):
     assert processError.value.e.returncode == 1
 
 
+def test_run_with_keyboard_interrupt(tmp_dir, tmp_venv, mocker):
+    mocker.patch("subprocess.run", side_effect=KeyboardInterrupt())
+    with pytest.raises(KeyboardInterrupt):
+        tmp_venv.run("python", "-", input_=MINIMAL_SCRIPT)
+    subprocess.run.assert_called_once()
+
+
+def test_call_with_input_and_keyboard_interrupt(tmp_dir, tmp_venv, mocker):
+    mocker.patch("subprocess.run", side_effect=KeyboardInterrupt())
+    kwargs = {"call": True}
+    with pytest.raises(KeyboardInterrupt):
+        tmp_venv.run("python", "-", input_=MINIMAL_SCRIPT, **kwargs)
+    subprocess.run.assert_called_once()
+
+
+def test_call_no_input_with_keyboard_interrupt(tmp_dir, tmp_venv, mocker):
+    mocker.patch("subprocess.call", side_effect=KeyboardInterrupt())
+    kwargs = {"call": True}
+    with pytest.raises(KeyboardInterrupt):
+        tmp_venv.run("python", "-", **kwargs)
+    subprocess.call.assert_called_once()
+
+
+def test_run_with_called_process_error(tmp_dir, tmp_venv, mocker):
+    mocker.patch(
+        "subprocess.run", side_effect=subprocess.CalledProcessError(42, "some_command")
+    )
+    with pytest.raises(EnvCommandError):
+        tmp_venv.run("python", "-", input_=MINIMAL_SCRIPT)
+    subprocess.run.assert_called_once()
+
+
+def test_call_with_input_and_called_process_error(tmp_dir, tmp_venv, mocker):
+    mocker.patch(
+        "subprocess.run", side_effect=subprocess.CalledProcessError(42, "some_command")
+    )
+    kwargs = {"call": True}
+    with pytest.raises(EnvCommandError):
+        tmp_venv.run("python", "-", input_=MINIMAL_SCRIPT, **kwargs)
+    subprocess.run.assert_called_once()
+
+
+def test_call_no_input_with_called_process_error(tmp_dir, tmp_venv, mocker):
+    mocker.patch(
+        "subprocess.call", side_effect=subprocess.CalledProcessError(42, "some_command")
+    )
+    kwargs = {"call": True}
+    with pytest.raises(EnvCommandError):
+        tmp_venv.run("python", "-", **kwargs)
+    subprocess.call.assert_called_once()
+
+
 def test_create_venv_tries_to_find_a_compatible_python_executable_using_generic_ones_first(
     manager, poetry, config, mocker, config_virtualenvs_path
 ):

tests/utils/test_pip.py

+import subprocess
+
+import pytest
+
+from poetry.utils.pip import pip_install
+
+
+def test_pip_install_successful(tmp_dir, tmp_venv, fixture_dir):
+    file_path = fixture_dir("distributions/demo-0.1.0-py2.py3-none-any.whl")
+    result = pip_install(file_path, tmp_venv)
+
+    assert "Successfully installed demo-0.1.0" in result
+
+
+def test_pip_install_with_keyboard_interrupt(tmp_dir, tmp_venv, fixture_dir, mocker):
+    file_path = fixture_dir("distributions/demo-0.1.0-py2.py3-none-any.whl")
+    mocker.patch("subprocess.run", side_effect=KeyboardInterrupt())
+    with pytest.raises(KeyboardInterrupt):
+        pip_install(file_path, tmp_venv)
+    subprocess.run.assert_called_once()