Prefer authenticator from repository when available

poetry/installation/chooser.py

@@ -9,6 +9,7 @@
 from poetry.core.packages.package import Package
 from poetry.core.packages.utils.link import Link
 from poetry.repositories.pool import Pool
+from poetry.repositories.repository import Repository
 from poetry.utils.env import Env
 from poetry.utils.patterns import wheel_file_re
 
@@ -79,7 +80,7 @@ def choose_for(self, package: Package) -> Link:
 
         return chosen
 
-    def _get_links(self, package: Package) -> List[Link]:
+    def get_repository(self, package: Package) -> Repository:
         if not package.source_type:
             if not self._pool.has_repository("pypi"):
                 repository = self._pool.repositories[0]
@@ -88,6 +89,10 @@ def _get_links(self, package: Package) -> List[Link]:
         else:
             repository = self._pool.repository(package.source_reference)
 
+        return repository
+
+    def _get_links(self, package: Package) -> List[Link]:
+        repository = self.get_repository(package)
         links = repository.find_links_for_package(package)
 
         hashes = [f["hash"] for f in package.files]

poetry/installation/executor.py

@@ -688,7 +688,9 @@ def _download_link(self, operation: Union[Install, Update], link: Link) -> Link:
         return archive
 
     def _download_archive(self, operation: Union[Install, Update], link: Link) -> Path:
-        response = self._authenticator.request(
+        repository = self._chooser.get_repository(operation.package)
+        authenticator = getattr(repository, "_authenticator", self._authenticator)
+        response = authenticator.request(
             "get", link.url, stream=True, io=self._sections.get(id(operation), self._io)
         )
         wheel_size = response.headers.get("content-length")

poetry/repositories/legacy_repository.py

@@ -28,7 +28,6 @@
 from poetry.core.semver.version_range import VersionRange
 from poetry.locations import REPOSITORY_CACHE_DIR
 from poetry.utils.helpers import canonicalize_name
-from poetry.utils.helpers import download_file
 from poetry.utils.helpers import temporary_directory
 from poetry.utils.patterns import wheel_file_re
 
@@ -444,6 +443,3 @@ def _get(self, endpoint: str) -> Optional[Page]:
             )
 
         return Page(response.url, response.content, response.headers)
-
-    def _download(self, url, dest):  # type: (str, str) -> None
-        return download_file(url, dest, session=self.session)

tests/installation/test_executor.py

@@ -21,7 +21,10 @@
 from poetry.installation.operations import Update
 from poetry.repositories.pool import Pool
 from poetry.utils.env import MockEnv
-from tests.repositories.test_pypi_repository import MockRepository
+from tests.repositories.test_legacy_repository import (
+    MockRepository as MockLegacyRepository,
+)
+from tests.repositories.test_pypi_repository import MockRepository as MockPypiRepository
 
 
 @pytest.fixture
@@ -62,7 +65,7 @@ def io_not_decorated():
 @pytest.fixture()
 def pool():
     pool = Pool()
-    pool.add_repository(MockRepository())
+    pool.add_repository(MockPypiRepository())
 
     return pool
 
@@ -429,6 +432,7 @@ def test_executor_should_write_pep610_url_references_for_urls(
         "demo",
         "0.1.0",
         source_type="url",
+        source_reference="pypi",
         source_url="https://files.pythonhosted.org/demo-0.1.0-py2.py3-none-any.whl",
     )
 
@@ -465,3 +469,31 @@ def test_executor_should_write_pep610_url_references_for_git(
             "url": package.source_url,
         },
     )
+
+
+@pytest.mark.parametrize("source_reference", ["pypi", "legacy"])
+def test_executor_should_use_repository_authenticator_when_available(
+    tmp_venv, pool, config, io, mock_file_downloads, http, source_reference
+):
+    legacy_repo = MockLegacyRepository()
+    legacy_repo._authenticator.session.headers["X-Sentinel"] = "test"
+    pool.add_repository(legacy_repo)
+
+    package = Package(
+        "demo",
+        "0.1.0",
+        source_type="url",
+        source_reference=source_reference,
+        source_url="https://files.pythonhosted.org/demo-0.1.0-py2.py3-none-any.whl",
+    )
+
+    executor = Executor(tmp_venv, pool, config, io)
+    rc = executor.execute([Install(package)])
+    assert rc == 0
+
+    req = http.last_request()
+
+    if source_reference == legacy_repo.name:
+        assert req.headers["X-Sentinel"] == "test"
+    else:
+        assert "X-Sentinel" not in req.headers