Merge pull request #9438 from sbidoul/legacy-cache-cleanup

Remove support for legacy cache entries

news/7502.removal.rst

@@ -0,0 +1,2 @@
+Remove support for legacy wheel cache entries that were created with pip
+versions older than 20.0.

src/pip/_internal/cache.py

@@ -55,34 +55,6 @@ def __init__(self, cache_dir, format_control, allowed_formats):
         _valid_formats = {"source", "binary"}
         assert self.allowed_formats.union(_valid_formats) == _valid_formats
 
-    def _get_cache_path_parts_legacy(self, link):
-        # type: (Link) -> List[str]
-        """Get parts of part that must be os.path.joined with cache_dir
-
-        Legacy cache key (pip < 20) for compatibility with older caches.
-        """
-
-        # We want to generate an url to use as our cache key, we don't want to
-        # just re-use the URL because it might have other items in the fragment
-        # and we don't care about those.
-        key_parts = [link.url_without_fragment]
-        if link.hash_name is not None and link.hash is not None:
-            key_parts.append("=".join([link.hash_name, link.hash]))
-        key_url = "#".join(key_parts)
-
-        # Encode our key url with sha224, we'll use this because it has similar
-        # security properties to sha256, but with a shorter total output (and
-        # thus less secure). However the differences don't make a lot of
-        # difference for our use case here.
-        hashed = hashlib.sha224(key_url.encode()).hexdigest()
-
-        # We want to nest the directories some to prevent having a ton of top
-        # level directories where we might run out of sub directories on some
-        # FS.
-        parts = [hashed[:2], hashed[2:4], hashed[4:6], hashed[6:]]
-
-        return parts
-
     def _get_cache_path_parts(self, link):
         # type: (Link) -> List[str]
         """Get parts of part that must be os.path.joined with cache_dir
@@ -139,17 +111,8 @@ def _get_candidates(self, link, canonical_package_name):
         if os.path.isdir(path):
             for candidate in os.listdir(path):
                 candidates.append((candidate, path))
-        # TODO remove legacy path lookup in pip>=21
-        legacy_path = self.get_path_for_link_legacy(link)
-        if os.path.isdir(legacy_path):
-            for candidate in os.listdir(legacy_path):
-                candidates.append((candidate, legacy_path))
         return candidates
 
-    def get_path_for_link_legacy(self, link):
-        # type: (Link) -> str
-        raise NotImplementedError()
-
     def get_path_for_link(self, link):
         # type: (Link) -> str
         """Return a directory to store cached items in for link.
@@ -177,12 +140,6 @@ def __init__(self, cache_dir, format_control):
         # type: (str, FormatControl) -> None
         super().__init__(cache_dir, format_control, {"binary"})
 
-    def get_path_for_link_legacy(self, link):
-        # type: (Link) -> str
-        parts = self._get_cache_path_parts_legacy(link)
-        assert self.cache_dir
-        return os.path.join(self.cache_dir, "wheels", *parts)
-
     def get_path_for_link(self, link):
         # type: (Link) -> str
         """Return a directory to store cached wheels for link
@@ -286,10 +243,6 @@ def __init__(self, cache_dir, format_control):
         self._wheel_cache = SimpleWheelCache(cache_dir, format_control)
         self._ephem_cache = EphemWheelCache(format_control)
 
-    def get_path_for_link_legacy(self, link):
-        # type: (Link) -> str
-        return self._wheel_cache.get_path_for_link_legacy(link)
-
     def get_path_for_link(self, link):
         # type: (Link) -> str
         return self._wheel_cache.get_path_for_link(link)

tests/unit/test_cache.py

@@ -51,47 +51,6 @@ def test_cache_hash():
     assert h == "f83b32dfa27a426dec08c21bf006065dd003d0aac78e7fc493d9014d"
 
 
-def test_get_path_for_link_legacy(tmpdir):
-    """
-    Test that an existing cache entry that was created with the legacy hashing
-    mechanism is returned by WheelCache._get_candidates().
-    """
-    wc = WheelCache(tmpdir, FormatControl())
-    link = Link("https://g.c/o/r")
-    path = wc.get_path_for_link(link)
-    legacy_path = wc.get_path_for_link_legacy(link)
-    assert path != legacy_path
-    ensure_dir(path)
-    with open(os.path.join(path, "test-1.0.0-pyz-none-any.whl"), "w"):
-        pass
-    ensure_dir(legacy_path)
-    with open(os.path.join(legacy_path, "test-1.0.0-pyx-none-any.whl"), "w"):
-        pass
-    expected_candidates = {
-        "test-1.0.0-pyx-none-any.whl", "test-1.0.0-pyz-none-any.whl"
-    }
-    candidates = {c[0] for c in wc._get_candidates(link, "test")}
-    assert candidates == expected_candidates
-
-
-def test_get_with_legacy_entry_only(tmpdir):
-    """
-    Test that an existing cache entry that was created with the legacy hashing
-    mechanism is actually returned in WheelCache.get().
-    """
-    wc = WheelCache(tmpdir, FormatControl())
-    link = Link("https://g.c/o/r")
-    legacy_path = wc.get_path_for_link_legacy(link)
-    ensure_dir(legacy_path)
-    with open(os.path.join(legacy_path, "test-1.0.0-py3-none-any.whl"), "w"):
-        pass
-    cached_link = wc.get(link, "test", [Tag("py3", "none", "any")])
-    assert (
-        os.path.normcase(os.path.dirname(cached_link.file_path)) ==
-        os.path.normcase(legacy_path)
-    )
-
-
 def test_get_cache_entry(tmpdir):
     wc = WheelCache(tmpdir, FormatControl())
     persi_link = Link("https://g.c/o/r/persi")