Remove pkg_resources usage in direct_url_helper

Author: uranusjr

src/pip/_internal/metadata/__init__.py

@@ -2,6 +2,14 @@
 
 from .base import BaseDistribution, BaseEnvironment
 
+__all__ = [
+    "BaseDistribution",
+    "BaseEnvironment",
+    "get_default_environment",
+    "get_environment",
+    "get_wheel_distribution",
+]
+
 
 def get_default_environment() -> BaseEnvironment:
     """Get the default representation for the current environment.

src/pip/_internal/metadata/base.py

@@ -1,4 +1,5 @@
 import email.message
+import json
 import logging
 import re
 from typing import (
@@ -15,7 +16,11 @@
 from pip._vendor.packaging.requirements import Requirement
 from pip._vendor.packaging.version import LegacyVersion, Version
 
-from pip._internal.models.direct_url import DirectUrl
+from pip._internal.models.direct_url import (
+    DIRECT_URL_METADATA_NAME,
+    DirectUrl,
+    DirectUrlValidationError,
+)
 from pip._internal.utils.misc import stdlib_pkgs  # TODO: Move definition here.
 
 if TYPE_CHECKING:
@@ -63,7 +68,29 @@ def version(self) -> DistributionVersion:
 
     @property
     def direct_url(self) -> Optional[DirectUrl]:
-        raise NotImplementedError()
+        """Obtain a DirectUrl from this distribution.
+
+        Returns None if the distribution has no `direct_url.json` metadata,
+        or if `direct_url.json` is invalid.
+        """
+        try:
+            content = self.read_text(DIRECT_URL_METADATA_NAME)
+        except FileNotFoundError:
+            return None
+        try:
+            return DirectUrl.from_json(content)
+        except (
+            UnicodeDecodeError,
+            json.JSONDecodeError,
+            DirectUrlValidationError,
+        ) as e:
+            logger.warning(
+                "Error parsing %s for %s: %s",
+                DIRECT_URL_METADATA_NAME,
+                self.canonical_name,
+                e,
+            )
+            return None
 
     @property
     def installer(self) -> str:
@@ -82,7 +109,11 @@ def in_usersite(self) -> bool:
         raise NotImplementedError()
 
     def read_text(self, name: str) -> str:
-        """Read a file in the .dist-info (or .egg-info) directory."""
+        """Read a file in the .dist-info (or .egg-info) directory.
+
+        Should raise ``FileNotFoundError`` if ``name`` does not exist in the
+        metadata directory.
+        """
         raise NotImplementedError()
 
     def iter_entry_points(self) -> Iterable[BaseEntryPoint]:

src/pip/_internal/metadata/pkg_resources.py

@@ -8,9 +8,7 @@
 from pip._vendor.packaging.utils import canonicalize_name
 from pip._vendor.packaging.version import parse as parse_version
 
-from pip._internal.models.direct_url import DirectUrl
 from pip._internal.utils import misc  # TODO: Move definition here.
-from pip._internal.utils.direct_url_helpers import dist_get_direct_url
 from pip._internal.utils.packaging import get_installer, get_metadata
 from pip._internal.utils.wheel import pkg_resources_distribution_for_wheel
 
@@ -47,10 +45,6 @@ def canonical_name(self) -> str:
     def version(self) -> DistributionVersion:
         return parse_version(self._dist.version)
 
-    @property
-    def direct_url(self) -> Optional[DirectUrl]:
-        return dist_get_direct_url(self._dist)
-
     @property
     def installer(self) -> str:
         return get_installer(self._dist)

src/pip/_internal/utils/direct_url_helpers.py

@@ -1,22 +1,9 @@
-import json
-import logging
 from typing import Optional
 
-from pip._vendor.pkg_resources import Distribution
-
-from pip._internal.models.direct_url import (
-    DIRECT_URL_METADATA_NAME,
-    ArchiveInfo,
-    DirectUrl,
-    DirectUrlValidationError,
-    DirInfo,
-    VcsInfo,
-)
+from pip._internal.models.direct_url import ArchiveInfo, DirectUrl, DirInfo, VcsInfo
 from pip._internal.models.link import Link
 from pip._internal.vcs import vcs
 
-logger = logging.getLogger(__name__)
-
 
 def direct_url_as_pep440_direct_reference(direct_url, name):
     # type: (DirectUrl, str) -> str
@@ -90,28 +77,3 @@ def direct_url_from_link(link, source_dir=None, link_is_in_wheel_cache=False):
             info=ArchiveInfo(hash=hash),
             subdirectory=link.subdirectory_fragment,
         )
-
-
-def dist_get_direct_url(dist):
-    # type: (Distribution) -> Optional[DirectUrl]
-    """Obtain a DirectUrl from a pkg_resource.Distribution.
-
-    Returns None if the distribution has no `direct_url.json` metadata,
-    or if `direct_url.json` is invalid.
-    """
-    if not dist.has_metadata(DIRECT_URL_METADATA_NAME):
-        return None
-    try:
-        return DirectUrl.from_json(dist.get_metadata(DIRECT_URL_METADATA_NAME))
-    except (
-        DirectUrlValidationError,
-        json.JSONDecodeError,
-        UnicodeDecodeError,
-    ) as e:
-        logger.warning(
-            "Error parsing %s for %s: %s",
-            DIRECT_URL_METADATA_NAME,
-            dist.project_name,
-            e,
-        )
-        return None

tests/unit/test_direct_url_helpers.py

@@ -1,18 +1,11 @@
 from functools import partial
-from unittest.mock import MagicMock, patch
-
-from pip._internal.models.direct_url import (
-    DIRECT_URL_METADATA_NAME,
-    ArchiveInfo,
-    DirectUrl,
-    DirInfo,
-    VcsInfo,
-)
+from unittest.mock import patch
+
+from pip._internal.models.direct_url import ArchiveInfo, DirectUrl, DirInfo, VcsInfo
 from pip._internal.models.link import Link
 from pip._internal.utils.direct_url_helpers import (
     direct_url_as_pep440_direct_reference,
     direct_url_from_link,
-    dist_get_direct_url,
 )
 from pip._internal.utils.urls import path_to_url
 
@@ -181,30 +174,3 @@ def test_from_link_hide_user_password():
         link_is_in_wheel_cache=True,
     )
     assert direct_url.to_dict()["url"] == "ssh://[email protected]/u/p.git"
-
-
-def test_dist_get_direct_url_no_metadata():
-    dist = MagicMock()
-    dist.has_metadata.return_value = False
-    assert dist_get_direct_url(dist) is None
-    dist.has_metadata.assert_called()
-
-
-def test_dist_get_direct_url_bad_metadata():
-    dist = MagicMock()
-    dist.has_metadata.return_value = True
-    dist.get_metadata.return_value = "{}"  # invalid direct_url.json
-    assert dist_get_direct_url(dist) is None
-    dist.get_metadata.assert_called_with(DIRECT_URL_METADATA_NAME)
-
-
-def test_dist_get_direct_url_valid_metadata():
-    dist = MagicMock()
-    dist.has_metadata.return_value = True
-    dist.get_metadata.return_value = (
-        '{"url": "https://e.c/p.tgz", "archive_info": {}}'
-    )
-    direct_url = dist_get_direct_url(dist)
-    dist.get_metadata.assert_called_with(DIRECT_URL_METADATA_NAME)
-    assert direct_url.url == "https://e.c/p.tgz"
-    assert isinstance(direct_url.info, ArchiveInfo)

tests/unit/test_metadata.py

@@ -0,0 +1,40 @@
+import logging
+from unittest.mock import patch
+
+from pip._internal.models.direct_url import DIRECT_URL_METADATA_NAME, ArchiveInfo
+from pip._internal.metadata import BaseDistribution
+
+
+@patch.object(BaseDistribution, "read_text", side_effect=FileNotFoundError)
+def test_dist_get_direct_url_no_metadata(mock_read_text):
+    dist = BaseDistribution()
+    assert dist.direct_url is None
+    mock_read_text.assert_called_once_with(DIRECT_URL_METADATA_NAME)
+
+
+@patch.object(BaseDistribution, "read_text", return_value="{}")
+def test_dist_get_direct_url_invalid_json(mock_read_text, caplog):
+    class FakeDistribution(BaseDistribution):
+        canonical_name = "whatever"  # Needed for error logging.
+
+    dist = FakeDistribution()
+    with caplog.at_level(logging.WARNING):
+        assert dist.direct_url is None
+
+    mock_read_text.assert_called_once_with(DIRECT_URL_METADATA_NAME)
+    assert caplog.records[-1].getMessage().startswith(
+        "Error parsing direct_url.json for whatever:",
+    )
+
+
+@patch.object(
+    BaseDistribution,
+    "read_text",
+    return_value='{"url": "https://e.c/p.tgz", "archive_info": {}}',
+)
+def test_dist_get_direct_url_valid_metadata(mock_read_text):
+    dist = BaseDistribution()
+    direct_url = dist.direct_url
+    mock_read_text.assert_called_once_with(DIRECT_URL_METADATA_NAME)
+    assert direct_url.url == "https://e.c/p.tgz"
+    assert isinstance(direct_url.info, ArchiveInfo)