Merge pull request #10675 from uranusjr/metadata-misc

Author: uranusjr

src/pip/_internal/exceptions.py

@@ -3,9 +3,8 @@
 import configparser
 import re
 from itertools import chain, groupby, repeat
-from typing import TYPE_CHECKING, Dict, Iterator, List, Optional, Union
+from typing import TYPE_CHECKING, Dict, Iterator, List, Optional
 
-from pip._vendor.pkg_resources import Distribution
 from pip._vendor.requests.models import Request, Response
 
 if TYPE_CHECKING:
@@ -149,17 +148,17 @@ def __init__(self, *, package: str, reason: str) -> None:
 
 
 class NoneMetadataError(PipError):
-    """
-    Raised when accessing "METADATA" or "PKG-INFO" metadata for a
-    pip._vendor.pkg_resources.Distribution object and
-    `dist.has_metadata('METADATA')` returns True but
-    `dist.get_metadata('METADATA')` returns None (and similarly for
-    "PKG-INFO").
+    """Raised when accessing a Distribution's "METADATA" or "PKG-INFO".
+
+    This signifies an inconsistency, when the Distribution claims to have
+    the metadata file (if not, raise ``FileNotFoundError`` instead), but is
+    not actually able to produce its content. This may be due to permission
+    errors.
     """
 
     def __init__(
         self,
-        dist: Union[Distribution, "BaseDistribution"],
+        dist: "BaseDistribution",
         metadata_name: str,
     ) -> None:
         """

src/pip/_internal/metadata/base.py

@@ -23,13 +23,19 @@
 from pip._vendor.packaging.utils import NormalizedName
 from pip._vendor.packaging.version import LegacyVersion, Version
 
+from pip._internal.exceptions import NoneMetadataError
+from pip._internal.locations import site_packages, user_site
 from pip._internal.models.direct_url import (
     DIRECT_URL_METADATA_NAME,
     DirectUrl,
     DirectUrlValidationError,
 )
 from pip._internal.utils.compat import stdlib_pkgs  # TODO: Move definition here.
-from pip._internal.utils.egg_link import egg_link_path_from_sys_path
+from pip._internal.utils.egg_link import (
+    egg_link_path_from_location,
+    egg_link_path_from_sys_path,
+)
+from pip._internal.utils.misc import is_local, normalize_path
 from pip._internal.utils.urls import url_to_path
 
 if TYPE_CHECKING:
@@ -131,6 +137,26 @@ def editable_project_location(self) -> Optional[str]:
                 return self.location
         return None
 
+    @property
+    def installed_location(self) -> Optional[str]:
+        """The distribution's "installed" location.
+
+        This should generally be a ``site-packages`` directory. This is
+        usually ``dist.location``, except for legacy develop-installed packages,
+        where ``dist.location`` is the source code location, and this is where
+        the ``.egg-link`` file is.
+
+        The returned location is normalized (in particular, with symlinks removed).
+        """
+        egg_link = egg_link_path_from_location(self.raw_name)
+        if egg_link:
+            location = egg_link
+        elif self.location:
+            location = self.location
+        else:
+            return None
+        return normalize_path(location)
+
     @property
     def info_location(self) -> Optional[str]:
         """Location of the .[egg|dist]-info directory or file.
@@ -250,23 +276,41 @@ def direct_url(self) -> Optional[DirectUrl]:
 
     @property
     def installer(self) -> str:
-        raise NotImplementedError()
+        try:
+            installer_text = self.read_text("INSTALLER")
+        except (OSError, ValueError, NoneMetadataError):
+            return ""  # Fail silently if the installer file cannot be read.
+        for line in installer_text.splitlines():
+            cleaned_line = line.strip()
+            if cleaned_line:
+                return cleaned_line
+        return ""
 
     @property
     def editable(self) -> bool:
         return bool(self.editable_project_location)
 
     @property
     def local(self) -> bool:
-        raise NotImplementedError()
+        """If distribution is installed in the current virtual environment.
+
+        Always True if we're not in a virtualenv.
+        """
+        if self.installed_location is None:
+            return False
+        return is_local(self.installed_location)
 
     @property
     def in_usersite(self) -> bool:
-        raise NotImplementedError()
+        if self.installed_location is None or user_site is None:
+            return False
+        return self.installed_location.startswith(normalize_path(user_site))
 
     @property
     def in_site_packages(self) -> bool:
-        raise NotImplementedError()
+        if self.installed_location is None or site_packages is None:
+            return False
+        return self.installed_location.startswith(normalize_path(site_packages))
 
     def is_file(self, path: InfoPath) -> bool:
         """Check whether an entry in the info directory is a file."""
@@ -286,6 +330,8 @@ def read_text(self, path: InfoPath) -> str:
         """Read a file in the info directory.
 
         :raise FileNotFoundError: If ``name`` does not exist in the directory.
+        :raise NoneMetadataError: If ``name`` exists in the info directory, but
+            cannot be read.
         """
         raise NotImplementedError()
 
@@ -294,7 +340,13 @@ def iter_entry_points(self) -> Iterable[BaseEntryPoint]:
 
     @property
     def metadata(self) -> email.message.Message:
-        """Metadata of distribution parsed from e.g. METADATA or PKG-INFO."""
+        """Metadata of distribution parsed from e.g. METADATA or PKG-INFO.
+
+        This should return an empty message if the metadata file is unavailable.
+
+        :raises NoneMetadataError: If the metadata file is available, but does
+            not contain valid metadata.
+        """
         raise NotImplementedError()
 
     @property
@@ -402,7 +454,11 @@ def from_paths(cls, paths: Optional[List[str]]) -> "BaseEnvironment":
         raise NotImplementedError()
 
     def get_distribution(self, name: str) -> Optional["BaseDistribution"]:
-        """Given a requirement name, return the installed distributions."""
+        """Given a requirement name, return the installed distributions.
+
+        The name may not be normalized. The implementation must canonicalize
+        it for lookup.
+        """
         raise NotImplementedError()
 
     def _iter_distributions(self) -> Iterator["BaseDistribution"]:

src/pip/_internal/metadata/pkg_resources.py

@@ -1,19 +1,19 @@
 import email.message
+import email.parser
 import logging
 import os
 import pathlib
-from typing import Collection, Iterable, Iterator, List, NamedTuple, Optional
-from zipfile import BadZipFile
+import zipfile
+from typing import Collection, Iterable, Iterator, List, Mapping, NamedTuple, Optional
 
 from pip._vendor import pkg_resources
 from pip._vendor.packaging.requirements import Requirement
 from pip._vendor.packaging.utils import NormalizedName, canonicalize_name
 from pip._vendor.packaging.version import parse as parse_version
 
-from pip._internal.exceptions import InvalidWheel
-from pip._internal.utils import misc  # TODO: Move definition here.
-from pip._internal.utils.packaging import get_installer, get_metadata
-from pip._internal.utils.wheel import pkg_resources_distribution_for_wheel
+from pip._internal.exceptions import InvalidWheel, NoneMetadataError, UnsupportedWheel
+from pip._internal.utils.misc import display_path
+from pip._internal.utils.wheel import parse_wheel, read_wheel_metadata_file
 
 from .base import (
     BaseDistribution,
@@ -33,6 +33,41 @@ class EntryPoint(NamedTuple):
     group: str
 
 
+class WheelMetadata:
+    """IMetadataProvider that reads metadata files from a dictionary.
+
+    This also maps metadata decoding exceptions to our internal exception type.
+    """
+
+    def __init__(self, metadata: Mapping[str, bytes], wheel_name: str) -> None:
+        self._metadata = metadata
+        self._wheel_name = wheel_name
+
+    def has_metadata(self, name: str) -> bool:
+        return name in self._metadata
+
+    def get_metadata(self, name: str) -> str:
+        try:
+            return self._metadata[name].decode()
+        except UnicodeDecodeError as e:
+            # Augment the default error with the origin of the file.
+            raise UnsupportedWheel(
+                f"Error decoding metadata for {self._wheel_name}: {e} in {name} file"
+            )
+
+    def get_metadata_lines(self, name: str) -> Iterable[str]:
+        return pkg_resources.yield_lines(self.get_metadata(name))
+
+    def metadata_isdir(self, name: str) -> bool:
+        return False
+
+    def metadata_listdir(self, name: str) -> List[str]:
+        return []
+
+    def run_script(self, script_name: str, namespace: str) -> None:
+        pass
+
+
 class Distribution(BaseDistribution):
     def __init__(self, dist: pkg_resources.Distribution) -> None:
         self._dist = dist
@@ -63,12 +98,26 @@ def from_wheel(cls, wheel: Wheel, name: str) -> "Distribution":
 
         :raises InvalidWheel: Whenever loading of the wheel causes a
             :py:exc:`zipfile.BadZipFile` exception to be thrown.
+        :raises UnsupportedWheel: If the wheel is a valid zip, but malformed
+            internally.
         """
         try:
             with wheel.as_zipfile() as zf:
-                dist = pkg_resources_distribution_for_wheel(zf, name, wheel.location)
-        except BadZipFile as e:
+                info_dir, _ = parse_wheel(zf, name)
+                metadata_text = {
+                    path.split("/", 1)[-1]: read_wheel_metadata_file(zf, path)
+                    for path in zf.namelist()
+                    if path.startswith(f"{info_dir}/")
+                }
+        except zipfile.BadZipFile as e:
             raise InvalidWheel(wheel.location, name) from e
+        except UnsupportedWheel as e:
+            raise UnsupportedWheel(f"{name} has an invalid wheel, {e}")
+        dist = pkg_resources.DistInfoDistribution(
+            location=wheel.location,
+            metadata=WheelMetadata(metadata_text, wheel.location),
+            project_name=name,
+        )
         return cls(dist)
 
     @property
@@ -97,25 +146,6 @@ def canonical_name(self) -> NormalizedName:
     def version(self) -> DistributionVersion:
         return parse_version(self._dist.version)
 
-    @property
-    def installer(self) -> str:
-        try:
-            return get_installer(self._dist)
-        except (OSError, ValueError):
-            return ""  # Fail silently if the installer file cannot be read.
-
-    @property
-    def local(self) -> bool:
-        return misc.dist_is_local(self._dist)
-
-    @property
-    def in_usersite(self) -> bool:
-        return misc.dist_in_usersite(self._dist)
-
-    @property
-    def in_site_packages(self) -> bool:
-        return misc.dist_in_site_packages(self._dist)
-
     def is_file(self, path: InfoPath) -> bool:
         return self._dist.has_metadata(str(path))
 
@@ -132,7 +162,10 @@ def read_text(self, path: InfoPath) -> str:
         name = str(path)
         if not self._dist.has_metadata(name):
             raise FileNotFoundError(name)
-        return self._dist.get_metadata(name)
+        content = self._dist.get_metadata(name)
+        if content is None:
+            raise NoneMetadataError(self, name)
+        return content
 
     def iter_entry_points(self) -> Iterable[BaseEntryPoint]:
         for group, entries in self._dist.get_entry_map().items():
@@ -142,7 +175,26 @@ def iter_entry_points(self) -> Iterable[BaseEntryPoint]:
 
     @property
     def metadata(self) -> email.message.Message:
-        return get_metadata(self._dist)
+        """
+        :raises NoneMetadataError: if the distribution reports `has_metadata()`
+            True but `get_metadata()` returns None.
+        """
+        if isinstance(self._dist, pkg_resources.DistInfoDistribution):
+            metadata_name = "METADATA"
+        else:
+            metadata_name = "PKG-INFO"
+        try:
+            metadata = self.read_text(metadata_name)
+        except FileNotFoundError:
+            if self.location:
+                displaying_path = display_path(self.location)
+            else:
+                displaying_path = repr(self.location)
+            logger.warning("No metadata found in %s", displaying_path)
+            metadata = ""
+        feed_parser = email.parser.FeedParser()
+        feed_parser.feed(metadata)
+        return feed_parser.close()
 
     def iter_dependencies(self, extras: Collection[str] = ()) -> Iterable[Requirement]:
         if extras:  # pkg_resources raises on invalid extras, so we sanitize.
@@ -178,7 +230,6 @@ def _search_distribution(self, name: str) -> Optional[BaseDistribution]:
         return None
 
     def get_distribution(self, name: str) -> Optional[BaseDistribution]:
-
         # Search the distribution by looking through the working set.
         dist = self._search_distribution(name)
         if dist: