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,25 @@ def from_dict(cls, d: Dict[str, Any]) -> "DirectUrl":
             ),
         )
 
-    def to_dict(self) -> Dict[str, Any]:
+    def to_dict(self, *, keep_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_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_hash_key: bool = True) -> str:
+        return json.dumps(self.to_dict(keep_hash_key=keep_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_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/unit/test_direct_url.py

@@ -28,6 +28,26 @@ def test_to_json() -> None:
     )
 
 
+def test_to_json_no_keep_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_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",

tests/unit/test_wheel.py

@@ -18,6 +18,7 @@
 from pip._internal.locations import get_scheme
 from pip._internal.models.direct_url import (
     DIRECT_URL_METADATA_NAME,
+    PROVENANCE_URL_METADATA_NAME,
     ArchiveInfo,
     DirectUrl,
 )
@@ -330,6 +331,17 @@ def main():
                 "gui_scripts": ["sample2 = sample:main"],
             },
         ).save_to_dir(tmpdir)
+        self.download_info = DirectUrl(
+            url="https://localhost:8080/sample/sample-1.2.0-py3-none-any.whl",
+            info=ArchiveInfo(
+                hash="sha256=257ded4ea1fafa475f099e544b2d7560f674d42917e096d462e"
+                "8a46a64f51245",
+                hashes={
+                    "sha256": "257ded4ea1fafa475f099e544b2d7560f674d42917e096d46"
+                    "2e8a46a64f51245",
+                },
+            ),
+        )
         self.req = Requirement("sample")
         self.src = os.path.join(tmpdir, "src")
         self.dest = os.path.join(tmpdir, "dest")
@@ -370,6 +382,8 @@ def test_std_install(self, data: TestData, tmpdir: Path) -> None:
             self.name,
             self.wheelpath,
             scheme=self.scheme,
+            download_info=self.download_info,
+            is_direct=False,
             req_description=str(self.req),
         )
         self.assert_installed(0o644)
@@ -388,6 +402,8 @@ def test_std_install_with_custom_umask(
                 self.name,
                 self.wheelpath,
                 scheme=self.scheme,
+                download_info=self.download_info,
+                is_direct=False,
                 req_description=str(self.req),
             )
             self.assert_installed(expected_permission)
@@ -400,6 +416,8 @@ def test_std_install_requested(self, data: TestData, tmpdir: Path) -> None:
             self.name,
             self.wheelpath,
             scheme=self.scheme,
+            download_info=self.download_info,
+            is_direct=False,
             req_description=str(self.req),
             requested=True,
         )
@@ -415,27 +433,100 @@ def test_std_install_with_direct_url(self, data: TestData, tmpdir: Path) -> None
         because wheelpath is typically the result of a local build.
         """
         self.prep(data, tmpdir)
-        direct_url = DirectUrl(
+        download_info = DirectUrl(
             url="file:///home/user/archive.tgz",
             info=ArchiveInfo(),
         )
         wheel.install_wheel(
             self.name,
             self.wheelpath,
             scheme=self.scheme,
+            download_info=download_info,
+            is_direct=True,
             req_description=str(self.req),
-            direct_url=direct_url,
         )
         direct_url_path = os.path.join(self.dest_dist_info, DIRECT_URL_METADATA_NAME)
         self.assert_permission(direct_url_path, 0o644)
         with open(direct_url_path, "rb") as f1:
-            expected_direct_url_json = direct_url.to_json()
+            expected_direct_url_json = download_info.to_json()
             direct_url_json = f1.read().decode("utf-8")
             assert direct_url_json == expected_direct_url_json
-        # check that the direc_url file is part of RECORDS
+        # check that the direct_url file is part of RECORDS
         with open(os.path.join(self.dest_dist_info, "RECORD")) as f2:
             assert DIRECT_URL_METADATA_NAME in f2.read()
 
+    def test_std_install_with_provenance_url(
+        self, data: TestData, tmpdir: Path
+    ) -> None:
+        """Test that install_wheel creates provenance_url.json metadata."""
+        self.prep(data, tmpdir)
+        download_info = 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",
+                },
+            ),
+        )
+        wheel.install_wheel(
+            self.name,
+            self.wheelpath,
+            scheme=self.scheme,
+            download_info=download_info,
+            is_direct=False,
+            req_description=str(self.req),
+        )
+        provenance_url_path = os.path.join(
+            self.dest_dist_info, PROVENANCE_URL_METADATA_NAME
+        )
+        self.assert_permission(provenance_url_path, 0o644)
+        with open(provenance_url_path, "rb") as f1:
+            expected_provenance_url_json = download_info.to_json(keep_hash_key=False)
+            provenance_url_json = f1.read().decode("utf-8")
+            assert provenance_url_json == expected_provenance_url_json
+        # check that the provenance_url.json file is part of RECORDS
+        with open(os.path.join(self.dest_dist_info, "RECORD")) as f2:
+            assert PROVENANCE_URL_METADATA_NAME in f2.read()
+
+    @pytest.mark.parametrize(
+        "hashes",
+        [
+            pytest.param(None, id="None"),
+            pytest.param({}, id="empty"),
+        ],
+    )
+    def test_std_install_with_provenance_url_no_hashes(
+        self, data: TestData, tmpdir: Path, hashes: Optional[Dict[str, str]]
+    ) -> None:
+        """Test that install_wheel does not create provenance_url.json
+        when hashes are missing.
+        """
+        self.prep(data, tmpdir)
+        download_info = DirectUrl(
+            url="https://pypi.org/simple/sample/sample-1.2.0-py3-none-any.whl",
+            info=ArchiveInfo(
+                hash=None,
+                hashes=hashes,
+            ),
+        )
+        wheel.install_wheel(
+            self.name,
+            self.wheelpath,
+            scheme=self.scheme,
+            download_info=download_info,
+            is_direct=False,
+            req_description=str(self.req),
+        )
+        provenance_url_path = os.path.join(
+            self.dest_dist_info, PROVENANCE_URL_METADATA_NAME
+        )
+        assert not os.path.exists(provenance_url_path)
+        with open(os.path.join(self.dest_dist_info, "RECORD")) as f2:
+            assert PROVENANCE_URL_METADATA_NAME not in f2.read()
+
     def test_install_prefix(self, data: TestData, tmpdir: Path) -> None:
         prefix = os.path.join(os.path.sep, "some", "path")
         self.prep(data, tmpdir)
@@ -451,6 +542,8 @@ def test_install_prefix(self, data: TestData, tmpdir: Path) -> None:
             self.name,
             self.wheelpath,
             scheme=scheme,
+            download_info=self.download_info,
+            is_direct=False,
             req_description=str(self.req),
         )
 
@@ -468,6 +561,8 @@ def test_dist_info_contains_empty_dir(self, data: TestData, tmpdir: Path) -> Non
             self.name,
             self.wheelpath,
             scheme=self.scheme,
+            download_info=self.download_info,
+            is_direct=False,
             req_description=str(self.req),
         )
         self.assert_installed(0o644)
@@ -486,6 +581,8 @@ def test_wheel_install_rejects_bad_paths(
                 "simple",
                 str(wheel_path),
                 scheme=self.scheme,
+                download_info=self.download_info,
+                is_direct=False,
                 req_description="simple",
             )
 
@@ -508,6 +605,8 @@ def test_invalid_entrypoints_fail(
                 "simple",
                 str(wheel_path),
                 scheme=self.scheme,
+                download_info=self.download_info,
+                is_direct=False,
                 req_description="simple",
             )