add mock server to test that each dist is downloaded exactly once

Author: cosmicexplorer

tests/conftest.py

@@ -992,3 +992,50 @@ def html_index_for_packages(
             f.write(pkg_index_content)
 
     return html_dir
+
+
+class OneTimeDownloadHandler(http.server.SimpleHTTPRequestHandler):
+    """Serve files from the current directory, but error if a file is downloaded more
+    than once."""
+
+    _seen_paths: ClassVar[Set[str]] = set()
+
+    def do_GET(self) -> None:
+        if self.path in self._seen_paths:
+            self.send_error(
+                http.HTTPStatus.NOT_FOUND,
+                f"File {self.path} not available more than once!",
+            )
+            return
+        super().do_GET()
+        if not (self.path.endswith("/") or self.path.endswith(".metadata")):
+            self._seen_paths.add(self.path)
+
+
+@pytest.fixture(scope="function")
+def html_index_with_onetime_server(
+    html_index_for_packages: Path,
+) -> Iterator[http.server.ThreadingHTTPServer]:
+    """Serve files from a generated pypi index, erroring if a file is downloaded more
+    than once.
+
+    Provide `-i http://localhost:8000` to pip invocations to point them at this server.
+    """
+    class InDirectoryServer(http.server.ThreadingHTTPServer):
+        def finish_request(self, request: Any, client_address: Any) -> None:
+            self.RequestHandlerClass(
+                request, client_address, self, directory=str(html_index_for_packages)  # type: ignore[call-arg] # noqa: E501
+            )
+
+    class Handler(OneTimeDownloadHandler):
+        _seen_paths: ClassVar[Set[str]] = set()
+
+    with InDirectoryServer(("", 8000), Handler) as httpd:
+        server_thread = threading.Thread(target=httpd.serve_forever)
+        server_thread.start()
+
+        try:
+            yield httpd
+        finally:
+            httpd.shutdown()
+            server_thread.join()

tests/functional/test_download.py

@@ -1324,6 +1324,57 @@ def test_download_metadata(
     assert sorted(os.listdir(download_dir)) == expected_outputs
 
 
+@pytest.mark.parametrize(
+    "requirement_to_download, expected_outputs, doubled_path",
+    [
+        (
+            "simple2==1.0",
+            ["simple-1.0.tar.gz", "simple2-1.0.tar.gz"],
+            "/simple2/simple2-1.0.tar.gz",
+        ),
+        ("simple==2.0", ["simple-2.0.tar.gz"], "/simple/simple-2.0.tar.gz"),
+        (
+            "colander",
+            ["colander-0.9.9-py2.py3-none-any.whl", "translationstring-1.1.tar.gz"],
+            "/colander/colander-0.9.9-py2.py3-none-any.whl",
+        ),
+        (
+            "compilewheel",
+            [
+                "compilewheel-1.0-py2.py3-none-any.whl",
+                "simple-1.0.tar.gz",
+            ],
+            "/compilewheel/compilewheel-1.0-py2.py3-none-any.whl",
+        ),
+    ],
+)
+def test_download_metadata_server(
+    download_server_html_index: Callable[..., Tuple[TestPipResult, Path]],
+    requirement_to_download: str,
+    expected_outputs: List[str],
+    doubled_path: str,
+) -> None:
+    """Verify that if a data-dist-info-metadata attribute is present, then it is used
+    instead of the actual dist's METADATA.
+
+    Additionally, verify that each dist is downloaded exactly once using a mock server.
+
+    This is a regression test for issue https://github.com/pypa/pip/issues/11847.
+    """
+    _, download_dir = download_server_html_index(
+        [requirement_to_download, "--no-cache-dir"],
+    )
+    assert sorted(os.listdir(download_dir)) == expected_outputs
+    shutil.rmtree(download_dir)
+    result, _ = download_server_html_index(
+        [requirement_to_download, "--no-cache-dir"],
+        allow_error=True,
+    )
+    assert result.returncode != 0
+    expected_msg = f"File {doubled_path} not available more than once!"
+    assert expected_msg in result.stderr
+
+
 @pytest.mark.parametrize(
     "requirement_to_download, real_hash",
     [