PEP-710: implement provenance_url.json file

Signed-off-by: Fridolin Pokorny <fridolin.pokorny@gmail.com>

Author: fridex

news/11865.feature.rst

@@ -0,0 +1 @@
+Implement PEP-710 for storing provenance_url.json file.

src/pip/_internal/models/direct_url.py

@@ -17,6 +17,7 @@
 T = TypeVar("T")
 
 DIRECT_URL_METADATA_NAME = "direct_url.json"
+PROVENANCE_URL_METADATA_NAME = "provenance_url.json"
 ENV_VAR_RE = re.compile(r"^\$\{[A-Za-z0-9-_]+\}(:\$\{[A-Za-z0-9-_]+\})?$")
 
 
@@ -205,20 +206,27 @@ def from_dict(cls, d: Dict[str, Any]) -> "DirectUrl":
             ),
         )
 
-    def to_dict(self) -> Dict[str, Any]:
+    def to_dict(self, *, keep_legacy_hash_key: bool = True) -> Dict[str, Any]:
         res = _filter_none(
             url=self.redacted_url,
             subdirectory=self.subdirectory,
         )
-        res[self.info.name] = self.info._to_dict()
+
+        info_dict = self.info._to_dict()
+        if not keep_legacy_hash_key:
+            info_dict.pop("hash", None)
+
+        res[self.info.name] = info_dict
         return res
 
     @classmethod
     def from_json(cls, s: str) -> "DirectUrl":
         return cls.from_dict(json.loads(s))
 
-    def to_json(self) -> str:
-        return json.dumps(self.to_dict(), sort_keys=True)
+    def to_json(self, *, keep_legacy_hash_key: bool = True) -> str:
+        return json.dumps(
+            self.to_dict(keep_legacy_hash_key=keep_legacy_hash_key), sort_keys=True
+        )
 
     def is_local_editable(self) -> bool:
         return isinstance(self.info, DirInfo) and self.info.editable

src/pip/_internal/operations/install/wheel.py

@@ -48,7 +48,12 @@
     FilesystemWheel,
     get_wheel_distribution,
 )
-from pip._internal.models.direct_url import DIRECT_URL_METADATA_NAME, DirectUrl
+from pip._internal.models.direct_url import (
+    DIRECT_URL_METADATA_NAME,
+    PROVENANCE_URL_METADATA_NAME,
+    ArchiveInfo,
+    DirectUrl,
+)
 from pip._internal.models.scheme import SCHEME_KEYS, Scheme
 from pip._internal.utils.filesystem import adjacent_tmp_file, replace
 from pip._internal.utils.misc import StreamWrapper, ensure_dir, hash_file, partition
@@ -424,9 +429,10 @@ def _install_wheel(  # noqa: C901, PLR0915 function is too long
     wheel_zip: ZipFile,
     wheel_path: str,
     scheme: Scheme,
+    download_info: DirectUrl,
+    is_direct: bool,
     pycompile: bool = True,
     warn_script_location: bool = True,
-    direct_url: Optional[DirectUrl] = None,
     requested: bool = False,
 ) -> None:
     """Install a wheel.
@@ -673,12 +679,25 @@ def _generate_file(path: str, **kwargs: Any) -> Generator[BinaryIO, None, None]:
         installer_file.write(b"pip\n")
     generated.append(installer_path)
 
-    # Record the PEP 610 direct URL reference
-    if direct_url is not None:
+    if is_direct:
+        # Record the PEP 610 direct URL reference
         direct_url_path = os.path.join(dest_info_dir, DIRECT_URL_METADATA_NAME)
         with _generate_file(direct_url_path) as direct_url_file:
-            direct_url_file.write(direct_url.to_json().encode("utf-8"))
+            direct_url_file.write(download_info.to_json().encode("utf-8"))
         generated.append(direct_url_path)
+    else:
+        # Record the PEP 710 provenance URL reference only if we have hashes for
+        # the given wheel. They can be missing when wheels are built using an old pip.
+        assert isinstance(download_info.info, ArchiveInfo)
+        if download_info.info.hashes:
+            provenance_url_path = os.path.join(
+                dest_info_dir, PROVENANCE_URL_METADATA_NAME
+            )
+            with _generate_file(provenance_url_path) as provenance_url_file:
+                provenance_url_file.write(
+                    download_info.to_json(keep_legacy_hash_key=False).encode("utf-8")
+                )
+            generated.append(provenance_url_path)
 
     # Record the REQUESTED file
     if requested:
@@ -721,10 +740,11 @@ def install_wheel(
     name: str,
     wheel_path: str,
     scheme: Scheme,
+    download_info: DirectUrl,
+    is_direct: bool,
     req_description: str,
     pycompile: bool = True,
     warn_script_location: bool = True,
-    direct_url: Optional[DirectUrl] = None,
     requested: bool = False,
 ) -> None:
     with ZipFile(wheel_path, allowZip64=True) as z:
@@ -734,8 +754,9 @@ def install_wheel(
                 wheel_zip=z,
                 wheel_path=wheel_path,
                 scheme=scheme,
+                download_info=download_info,
+                is_direct=is_direct,
                 pycompile=pycompile,
                 warn_script_location=warn_script_location,
-                direct_url=direct_url,
                 requested=requested,
             )

src/pip/_internal/req/req_install.py

@@ -861,17 +861,19 @@ def install(
             self.install_succeeded = True
             return
 
+        assert self.download_info
         assert self.is_wheel
         assert self.local_file_path
 
         install_wheel(
             self.req.name,
             self.local_file_path,
             scheme=scheme,
+            download_info=self.download_info,
+            is_direct=self.is_direct,
             req_description=str(self.req),
             pycompile=pycompile,
             warn_script_location=warn_script_location,
-            direct_url=self.download_info if self.is_direct else None,
             requested=self.user_supplied,
         )
         self.install_succeeded = True

tests/functional/test_install.py

@@ -1,5 +1,6 @@
 import hashlib
 import io
+import json
 import os
 import re
 import ssl
@@ -9,7 +10,7 @@
 import textwrap
 from os.path import curdir, join, pardir
 from pathlib import Path
-from typing import Dict, Iterable, List, Optional, Tuple
+from typing import Any, Dict, Iterable, List, Optional, Tuple
 
 import pytest
 
@@ -2743,3 +2744,154 @@ def add_link(tar: tarfile.TarFile, name: str, linktype: str, target: str) -> Non
     # Run the internal test
     result = script.run("python", "-m", "linktest")
     assert result.stdout.strip() == "8 files checked"
+
+
+def _check_provenance_url(provenance_url: Dict[str, Any]) -> None:
+    assert "archive_info" in provenance_url
+    assert "url" in provenance_url
+    assert (
+        len(provenance_url) == 2
+    ), "provenance_url.json should hold only archive_info and url keys"
+
+    assert "hashes" in provenance_url["archive_info"]
+    assert len(provenance_url["archive_info"]["hashes"]) > 0
+
+
+@pytest.mark.parametrize(
+    "pkg_name, pkg_version, distribution",
+    [
+        pytest.param(
+            "simplewheel",
+            "1.0",
+            "simplewheel-1.0-py2.py3-none-any.whl",
+            id="wheel",
+        ),
+        pytest.param(
+            "simple",
+            "1.0",
+            "simple-1.0.tar.gz",
+            id="sdist",
+        ),
+    ],
+)
+def test_install_provenance_url(
+    script: PipTestEnvironment,
+    data: TestData,
+    pkg_name: str,
+    pkg_version: str,
+    distribution: str,
+) -> None:
+    """Test installing a distribution from a simple API produces provenance_url.json."""
+    server = make_mock_server()
+
+    distribution_path = f"/files/{distribution}"
+    server.mock.side_effect = [
+        package_page(
+            {
+                distribution: distribution_path,
+            }
+        ),
+        file_response(data.packages.joinpath(distribution)),
+    ]
+
+    index_url = f"http://{server.host}:{server.port}"
+
+    pip_args = [
+        "install",
+        "-i",
+        index_url,
+        f"{pkg_name}=={pkg_version}",
+    ]
+    with server_running(server):
+        result = script.pip(*pip_args)
+
+        result.assert_installed(
+            pkg_name=pkg_name, without_egg_link=True, editable=False
+        )
+
+        provenance_url_path = (
+            script.site_packages
+            / f"{pkg_name}-{pkg_version}.dist-info"
+            / "provenance_url.json"
+        )
+
+        assert result.files_created[
+            provenance_url_path
+        ], "provenance_url.json was not created"
+
+        provenance_url_full_path = result.files_created[provenance_url_path].full
+
+        with open(provenance_url_full_path) as f:
+            provenance_url_content = json.load(f)
+
+        _check_provenance_url(provenance_url_content)
+        assert provenance_url_content["url"] == f"{index_url}{distribution_path}"
+
+
+def test_install_provenance_url_cached(
+    script: PipTestEnvironment, data: TestData
+) -> None:
+    """Test installing a cached distribution produced provenance_url.json."""
+    pkg_name = "simple"
+    pkg_version = "1.0"
+    distribution = "simple-1.0.tar.gz"
+
+    server = make_mock_server()
+
+    distribution_path = f"/files/{distribution}"
+    server.mock.side_effect = [
+        package_page(
+            {
+                distribution: distribution_path,
+            }
+        ),
+        file_response(data.packages.joinpath(distribution)),
+    ] * 2
+
+    index_url = f"http://{server.host}:{server.port}"
+
+    pip_args = [
+        "install",
+        "-i",
+        index_url,
+        f"{pkg_name}=={pkg_version}",
+    ]
+
+    with server_running(server):
+        result = script.pip(*pip_args)
+
+        result.assert_installed(
+            pkg_name=pkg_name, without_egg_link=True, editable=False
+        )
+
+        provenance_url_path = (
+            script.site_packages
+            / f"{pkg_name}-{pkg_version}.dist-info"
+            / "provenance_url.json"
+        )
+
+        assert result.files_created[
+            provenance_url_path
+        ], "provenance_url.json was not created"
+
+        provenance_url_full_path = result.files_created[provenance_url_path].full
+
+        with open(provenance_url_full_path) as f:
+            provenance_url_content = json.load(f)
+
+        _check_provenance_url(provenance_url_content)
+        assert provenance_url_content["url"] == f"{index_url}{distribution_path}"
+
+        os.unlink(provenance_url_full_path)
+
+        pip_args.append("--ignore-installed")
+        result = script.pip(*pip_args)
+
+        assert f"Using cached {pkg_name}" in result.stdout
+
+        assert os.path.exists(provenance_url_full_path)
+        with open(provenance_url_full_path) as f:
+            new_provenance_url_content = json.load(f)
+
+        _check_provenance_url(new_provenance_url_content)
+        assert new_provenance_url_content == provenance_url_content

tests/unit/test_direct_url.py

@@ -28,6 +28,26 @@ def test_to_json() -> None:
     )
 
 
+def test_to_json_no_keep_legacy_hash_key() -> None:
+    direct_url = DirectUrl(
+        url="https://pypi.org/simple/sample/sample-1.2.0-py3-none-any.whl",
+        info=ArchiveInfo(
+            hash="sha256=257ded4ea1fafa475f099e544b2d7560f674d42"
+            "917e096d462e8a46a64f51245",
+            hashes={
+                "sha256": "257ded4ea1fafa475f099e544b2d7560f674d"
+                "42917e096d462e8a46a64f51245",
+            },
+        ),
+    )
+    direct_url.validate()
+    assert direct_url.to_json(keep_legacy_hash_key=False) == (
+        '{"archive_info": {"hashes": {'
+        '"sha256": "257ded4ea1fafa475f099e544b2d7560f674d42917e096d462e8a46a64f51245"}'
+        '}, "url": "https://pypi.org/simple/sample/sample-1.2.0-py3-none-any.whl"}'
+    )
+
+
 def test_archive_info() -> None:
     direct_url_dict = {
         "url": "file:///home/user/archive.tgz",