installer: print warning if yanked file is used for install

src/poetry/installation/executor.py

@@ -78,6 +78,7 @@ def __init__(
         self._executed = {"install": 0, "update": 0, "uninstall": 0}
         self._skipped = {"install": 0, "update": 0, "uninstall": 0}
         self._sections: dict[int, SectionOutput] = {}
+        self._yanked_warnings: list[str] = []
         self._lock = threading.Lock()
         self._shutdown = False
         self._hashes: dict[str, str] = {}
@@ -140,6 +141,7 @@ def execute(self, operations: list[Operation]) -> int:
         # We group operations by priority
         groups = itertools.groupby(operations, key=lambda o: -o.priority)
         self._sections = {}
+        self._yanked_warnings = []
         for _, group in groups:
             tasks = []
             serial_operations = []
@@ -179,6 +181,9 @@ def execute(self, operations: list[Operation]) -> int:
 
                 break
 
+        for warning in self._yanked_warnings:
+            self._io.write_error_line(f"<warning>Warning: {warning}</warning>")
+
         return 1 if self._shutdown else 0
 
     @staticmethod
@@ -611,6 +616,18 @@ def _install_git(self, operation: Install | Update) -> int:
     def _download(self, operation: Install | Update) -> Path:
         link = self._chooser.choose_for(operation.package)
 
+        if link.yanked:
+            # Store yanked warnings in a list and print after installing, so they can't
+            # be overlooked. Further, printing them in the concerning section would have
+            # the risk of overwriting the warning, so it is only briefly visible.
+            message = (
+                f"The file chosen for install of {operation.package.pretty_name} "
+                f"{operation.package.pretty_version} ({link.show_url}) is yanked."
+            )
+            if link.yanked_reason:
+                message += f" Reason for being yanked: {link.yanked_reason}"
+            self._yanked_warnings.append(message)
+
         return self._download_link(operation, link)
 
     def _download_link(self, operation: Install | Update, link: Link) -> Path:

tests/installation/test_executor.py

@@ -32,6 +32,7 @@
     from pytest_mock import MockerFixture
 
     from poetry.config.config import Config
+    from poetry.installation.operations.operation import Operation
     from poetry.utils.env import VirtualEnv
     from tests.types import FixtureDirGetter
 
@@ -177,6 +178,61 @@ def test_execute_executes_a_batch_of_operations(
     assert pip_install.call_args.kwargs.get("editable", False)
 
 
+@pytest.mark.parametrize(
+    "operations, has_warning",
+    [
+        (
+            [Install(Package("black", "21.11b0")), Install(Package("pytest", "3.5.2"))],
+            True,
+        ),
+        (
+            [
+                Uninstall(Package("black", "21.11b0")),
+                Uninstall(Package("pytest", "3.5.2")),
+            ],
+            False,
+        ),
+        (
+            [
+                Update(Package("black", "19.10b0"), Package("black", "21.11b0")),
+                Update(Package("pytest", "3.5.1"), Package("pytest", "3.5.2")),
+            ],
+            True,
+        ),
+    ],
+)
+def test_execute_prints_warning_for_yanked_package(
+    config: Config,
+    pool: Pool,
+    io: BufferedIO,
+    tmp_dir: str,
+    mock_file_downloads: None,
+    env: MockEnv,
+    operations: list[Operation],
+    has_warning: bool,
+):
+    config.merge({"cache-dir": tmp_dir})
+
+    executor = Executor(env, pool, config, io)
+
+    return_code = executor.execute(operations)
+
+    expected = (
+        "Warning: The file chosen for install of black 21.11b0 "
+        "(black-21.11b0-py3-none-any.whl) is yanked. Reason for being yanked: "
+        "Broken regex dependency. Use 21.11b1 instead."
+    )
+    error = io.fetch_error()
+    assert return_code == 0
+    assert "pytest" not in error
+    if has_warning:
+        assert expected in error
+        assert error.count("is yanked") == 1
+    else:
+        assert expected not in error
+        assert error.count("yanked") == 0
+
+
 def test_execute_shows_skipped_operations_if_verbose(
     config: Config, pool: Pool, io: BufferedIO, config_cache_dir: Path, env: MockEnv
 ):