Add WheelCache method to inform which cache was used

Return whether the link was found in the persistent or ephemeral cache.
This commit is contained in:
Stéphane Bidoul 2020-02-01 13:40:25 +01:00
parent f77944733d
commit 94b77130aa
No known key found for this signature in database
GPG Key ID: BCAB2555446B5B92
2 changed files with 55 additions and 2 deletions

View File

@ -270,6 +270,16 @@ class EphemWheelCache(SimpleWheelCache):
)
class CacheEntry(object):
def __init__(
self,
link, # type: Link
persistent, # type: bool
):
self.link = link
self.persistent = persistent
class WheelCache(Cache):
"""Wraps EphemWheelCache and SimpleWheelCache into a single Cache
@ -304,16 +314,36 @@ class WheelCache(Cache):
supported_tags, # type: List[Tag]
):
# type: (...) -> Link
cache_entry = self.get_cache_entry(link, package_name, supported_tags)
if cache_entry is None:
return link
return cache_entry.link
def get_cache_entry(
self,
link, # type: Link
package_name, # type: Optional[str]
supported_tags, # type: List[Tag]
):
# type: (...) -> Optional[CacheEntry]
"""Returns a CacheEntry with a link to a cached item if it exists or
None. The cache entry indicates if the item was found in the persistent
or ephemeral cache.
"""
retval = self._wheel_cache.get(
link=link,
package_name=package_name,
supported_tags=supported_tags,
)
if retval is not link:
return retval
return CacheEntry(retval, persistent=True)
return self._ephem_cache.get(
retval = self._ephem_cache.get(
link=link,
package_name=package_name,
supported_tags=supported_tags,
)
if retval is not link:
return CacheEntry(retval, persistent=False)
return None

View File

@ -90,3 +90,26 @@ def test_get_with_legacy_entry_only(tmpdir):
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")
persi_path = wc.get_path_for_link(persi_link)
ensure_dir(persi_path)
with open(os.path.join(persi_path, "persi-1.0.0-py3-none-any.whl"), "w"):
pass
ephem_link = Link("https://g.c/o/r/ephem")
ephem_path = wc.get_ephem_path_for_link(ephem_link)
ensure_dir(ephem_path)
with open(os.path.join(ephem_path, "ephem-1.0.0-py3-none-any.whl"), "w"):
pass
other_link = Link("https://g.c/o/r/other")
supported_tags = [Tag("py3", "none", "any")]
assert (
wc.get_cache_entry(persi_link, "persi", supported_tags).persistent
)
assert (
not wc.get_cache_entry(ephem_link, "ephem", supported_tags).persistent
)
assert wc.get_cache_entry(other_link, "other", supported_tags) is None