Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@ classifiers = [
"Development Status :: 5 - Production/Stable",
]
requires-python = ">= 3.10"
dependencies = ["repoze.lru"]

[project.urls]
Documentation = "https://reg.readthedocs.io"
Expand Down Expand Up @@ -84,6 +83,7 @@ python_version = "3.10"
strict = true
files = ["."]
warn_unreachable = true
warn_unused_configs = true
warn_unused_ignores = true

[tool.pyright]
Expand Down
48 changes: 12 additions & 36 deletions reg/cache.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
from __future__ import annotations

from functools import lru_cache
from typing import TYPE_CHECKING, Any, Generic

from repoze.lru import lru_cache # type: ignore[import-untyped]

if TYPE_CHECKING:
from collections.abc import Callable, Sequence
from typing_extensions import TypeVar
Expand Down Expand Up @@ -49,27 +48,13 @@ class DictCachingKeyLookup(Generic[_ValueT]):

def __init__(self, key_lookup: KeyLookup[_ValueT]) -> None:
self.key_lookup = key_lookup
self.component = Cache(key_lookup.component).__getitem__ # type: ignore
self.fallback = Cache(key_lookup.fallback).__getitem__ # type: ignore
self.component = Cache(key_lookup.component).__getitem__
self.fallback = Cache(key_lookup.fallback).__getitem__

def _all(key: Sequence[Any]) -> list[_ValueT]:
return list(key_lookup.all(key))

self.all = Cache(_all).__getitem__ # type: ignore

if TYPE_CHECKING:
# NOTE: For pyright's sake we declare these callable instance attributes
# as methods, even though they're not, since pyright does not seem
# to be able to match protocols against them. mypy can deal with
# it just fine
def component(self, key: Sequence[Any], /) -> _ValueT | None:
raise NotImplementedError

def fallback(self, key: Sequence[Any], /) -> _ValueT | None:
raise NotImplementedError

def all(self, key: Sequence[Any], /) -> list[_ValueT]:
raise NotImplementedError
self.all = Cache(_all).__getitem__


class LruCachingKeyLookup(Generic[_ValueT]):
Expand Down Expand Up @@ -100,20 +85,11 @@ def __init__(
fallback_cache_size: int,
) -> None:
self.key_lookup = key_lookup
self.component = lru_cache(component_cache_size)(key_lookup.component) # type: ignore
self.fallback = lru_cache(fallback_cache_size)(key_lookup.fallback) # type: ignore
self.all = lru_cache(all_cache_size)(lambda key: list(key_lookup.all(key))) # type: ignore

if TYPE_CHECKING:
# NOTE: For pyright's sake we declare these callable instance attributes
# as methods, even though they're not, since pyright does not seem
# to be able to match protocols against them. mypy can deal with
# it just fine
def component(self, key: Sequence[Any], /) -> _ValueT | None:
raise NotImplementedError

def fallback(self, key: Sequence[Any], /) -> _ValueT | None:
raise NotImplementedError

def all(self, key: Sequence[Any], /) -> list[_ValueT]:
raise NotImplementedError
self.component = lru_cache(component_cache_size)(key_lookup.component)
self.fallback = lru_cache(fallback_cache_size)(key_lookup.fallback)

@lru_cache(all_cache_size)
def _all(key: Sequence[Any]) -> list[_ValueT]:
return list(key_lookup.all(key))

self.all = _all
195 changes: 195 additions & 0 deletions reg/tests/test_cache.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,195 @@
"""Tests for the caching key lookups in reg.cache.

These tests treat the caches as black boxes: a fake key lookup records every
call it receives, so we can assert how often the underlying functions were
really invoked. Nothing here depends on the cache implementation, only on
caching behavior.

The fake returns ``str`` values, so the lookups under test are typed as
``LruCachingKeyLookup[str]`` / ``DictCachingKeyLookup[str]``.
"""

from __future__ import annotations

from typing import TYPE_CHECKING, Any

from reg.cache import DictCachingKeyLookup, LruCachingKeyLookup

if TYPE_CHECKING:
from collections.abc import Iterator, Sequence


class FakeKeyLookup:
"""Records every call so tests can see what reached the underlying lookup."""

def __init__(self) -> None:
self.calls: list[tuple[str, Sequence[Any]]] = []

def component(self, key: Sequence[Any]) -> str | None:
self.calls.append(("component", key))
return f"component:{key}"

def fallback(self, key: Sequence[Any]) -> str | None:
self.calls.append(("fallback", key))
return None # None results must be cached too

def all(self, key: Sequence[Any]) -> Iterator[str]:
self.calls.append(("all", key))
return iter([f"all:{key}"]) # an iterator; the cache must turn it into a list


def make(
component: int = 10, all: int = 10, fallback: int = 10
) -> tuple[FakeKeyLookup, LruCachingKeyLookup[str]]:
kl = FakeKeyLookup()
return kl, LruCachingKeyLookup(kl, component, all, fallback)


# --- LruCachingKeyLookup: component ---------------------------------------


def test_component_result_is_returned() -> None:
_kl, cached = make()
assert cached.component(("a",)) == "component:('a',)"


def test_component_is_cached() -> None:
kl, cached = make()
cached.component(("a",))
cached.component(("a",))
assert kl.calls == [("component", ("a",))]


def test_component_different_keys_are_cached_separately() -> None:
kl, cached = make()
cached.component(("a",))
cached.component(("b",))
cached.component(("a",))
cached.component(("b",))
assert kl.calls == [("component", ("a",)), ("component", ("b",))]


# --- LruCachingKeyLookup: fallback ----------------------------------------


def test_fallback_is_cached() -> None:
kl, cached = make()
cached.fallback(("a",))
cached.fallback(("a",))
assert kl.calls == [("fallback", ("a",))]


def test_none_result_is_cached() -> None:
kl, cached = make()
assert cached.fallback(("a",)) is None
assert cached.fallback(("a",)) is None
assert len(kl.calls) == 1


# --- LruCachingKeyLookup: all ---------------------------------------------


def test_all_returns_a_list() -> None:
_kl, cached = make()
result = cached.all(("a",))
assert isinstance(result, list)
assert result == ["all:('a',)"]


def test_all_is_cached() -> None:
kl, cached = make()
first = cached.all(("a",))
second = cached.all(("a",))
assert len(kl.calls) == 1
assert first == second


# --- LruCachingKeyLookup: independence ------------------------------------


def test_same_key_is_cached_per_method() -> None:
kl, cached = make()
cached.component(("a",))
cached.fallback(("a",))
cached.all(("a",))
cached.component(("a",))
cached.fallback(("a",))
cached.all(("a",))
assert kl.calls == [
("component", ("a",)),
("fallback", ("a",)),
("all", ("a",)),
]


def test_cache_sizes_are_independent() -> None:
kl, cached = make(component=1, fallback=3)
cached.component(("a",))
cached.component(("b",)) # evicts "a" from the component cache only
for k in ("a", "b", "c"):
cached.fallback((k,))
kl.calls.clear()

cached.component(("a",))
for k in ("a", "b", "c"):
cached.fallback((k,))
assert kl.calls == [("component", ("a",))]


# --- LruCachingKeyLookup: eviction ----------------------------------------


def test_component_least_recently_used_is_evicted() -> None:
kl, cached = make(component=2)
cached.component(("a",))
cached.component(("b",))
cached.component(("c",)) # evicts "a"
kl.calls.clear()

cached.component(("b",)) # still cached
cached.component(("c",)) # still cached
assert kl.calls == []

cached.component(("a",)) # was evicted, recomputed
assert kl.calls == [("component", ("a",))]


def test_component_recent_use_protects_from_eviction() -> None:
kl, cached = make(component=2)
cached.component(("a",))
cached.component(("b",))
cached.component(("a",)) # "b" is now the least recently used
cached.component(("c",)) # evicts "b"
kl.calls.clear()

cached.component(("a",))
assert kl.calls == []
cached.component(("b",))
assert kl.calls == [("component", ("b",))]


# --- DictCachingKeyLookup (unbounded) -------------------------------------


def test_dict_component_is_cached() -> None:
kl = FakeKeyLookup()
cached: DictCachingKeyLookup[str] = DictCachingKeyLookup(kl)
assert cached.component(("a",)) == "component:('a',)"
cached.component(("a",))
assert kl.calls == [("component", ("a",))]


def test_dict_fallback_caches_none() -> None:
kl = FakeKeyLookup()
cached: DictCachingKeyLookup[str] = DictCachingKeyLookup(kl)
assert cached.fallback(("a",)) is None
assert cached.fallback(("a",)) is None
assert kl.calls == [("fallback", ("a",))]


def test_dict_all_returns_list_and_is_cached() -> None:
kl = FakeKeyLookup()
cached: DictCachingKeyLookup[str] = DictCachingKeyLookup(kl)
assert cached.all(("a",)) == ["all:('a',)"]
cached.all(("a",))
assert kl.calls == [("all", ("a",))]
Loading
Loading