import unittest import warnings from cachetools import Cache, cachedmethod, keys from . import CountedCondition, CountedLock class Cached: def __init__(self, cache, count=0): self.cache = cache self.count = count self.lock = CountedLock() self.cond = CountedCondition() def __get(self, _value): result = self.count self.count += 1 return result @cachedmethod(lambda self: self.cache) def get(self, value): """docstring""" return self.__get(value) @cachedmethod(lambda self: self.cache, key=keys.typedmethodkey) def get_typed(self, value): return self.__get(value) @cachedmethod(lambda self: self.cache, info=True) def get_info(self, value): return self.__get(value) @cachedmethod(lambda self: self.cache, lock=lambda self: self.lock) def get_lock(self, value): return self.__get(value) @cachedmethod(lambda self: self.cache, lock=lambda self: self.lock, info=True) def get_lock_info(self, value): return self.__get(value) @cachedmethod(lambda self: self.cache, condition=lambda self: self.cond) def get_cond(self, value): return self.__get(value) @cachedmethod(lambda self: self.cache, condition=lambda self: self.cond) def get_cond_error(self, _value): raise ValueError("test error") @cachedmethod(lambda self: self.cache, condition=lambda self: self.cond, info=True) def get_cond_info(self, value): return self.__get(value) @cachedmethod( lambda self: self.cache, lock=lambda self: self.lock, condition=lambda self: self.cond, ) def get_lock_cond(self, value): return self.__get(value) @cachedmethod( lambda self: self.cache, lock=lambda self: self.lock, condition=lambda self: self.cond, info=True, ) def get_lock_cond_info(self, value): return self.__get(value) class Unhashable(Cached): # https://github.com/tkem/cachetools/issues/107 def __hash__(self): raise TypeError("unhashable type") class MethodDecoratorTestMixin: def cache(self, _minsize): raise NotImplementedError def test_decorator(self): cached = Cached(self.cache(2)) self.assertEqual(cached.get(0), 0) self.assertEqual(cached.get(1), 1) self.assertEqual(cached.get(1), 1) self.assertEqual(cached.get(1.0), 1) self.assertEqual(cached.get(1.0), 1) cached.cache.clear() self.assertEqual(cached.get(1), 2) def test_decorator_typed(self): cached = Cached(self.cache(3)) self.assertEqual(cached.get_typed(0), 0) self.assertEqual(cached.get_typed(1), 1) self.assertEqual(cached.get_typed(1), 1) self.assertEqual(cached.get_typed(1.0), 2) self.assertEqual(cached.get_typed(1.0), 2) self.assertEqual(cached.get_typed(0.0), 3) def test_decorator_unhashable(self): cached = Unhashable(self.cache(2)) self.assertEqual(cached.get(0), 0) self.assertEqual(cached.get(1), 1) self.assertEqual(cached.get(1), 1) self.assertEqual(cached.get(1.0), 1) self.assertEqual(cached.get(1.0), 1) cached.cache.clear() self.assertEqual(cached.get(1), 2) def test_decorator_lock(self): cached = Cached(self.cache(2)) self.assertEqual(cached.get_lock(0), 0) self.assertEqual(cached.lock.count, 2) self.assertEqual(cached.get_lock(1), 1) self.assertEqual(cached.lock.count, 4) self.assertEqual(cached.get_lock(1), 1) self.assertEqual(cached.lock.count, 5) self.assertEqual(cached.get_lock(1.0), 1) self.assertEqual(cached.lock.count, 6) self.assertEqual(cached.get_lock(1.0), 1) self.assertEqual(cached.lock.count, 7) cached.cache.clear() self.assertEqual(cached.get_lock(1), 2) self.assertEqual(cached.lock.count, 9) def test_decorator_cond(self): cached = Cached(self.cache(2)) self.assertEqual(cached.get_cond(0), 0) self.assertEqual(cached.cond.count, 3) self.assertEqual(cached.cond.wait_count, 1) self.assertEqual(cached.cond.notify_count, 1) self.assertEqual(cached.get_cond(1), 1) self.assertEqual(cached.cond.count, 6) self.assertEqual(cached.cond.wait_count, 2) self.assertEqual(cached.cond.notify_count, 2) self.assertEqual(cached.get_cond(1), 1) self.assertEqual(cached.cond.count, 7) self.assertEqual(cached.cond.wait_count, 3) self.assertEqual(cached.cond.notify_count, 2) self.assertEqual(cached.get_cond(1.0), 1) self.assertEqual(cached.cond.count, 8) self.assertEqual(cached.cond.wait_count, 4) self.assertEqual(cached.cond.notify_count, 2) self.assertEqual(cached.get_cond(1.0), 1) self.assertEqual(cached.cond.count, 9) self.assertEqual(cached.cond.wait_count, 5) self.assertEqual(cached.cond.notify_count, 2) cached.cache.clear() self.assertEqual(cached.get_cond(1), 2) self.assertEqual(cached.cond.count, 12) self.assertEqual(cached.cond.wait_count, 6) self.assertEqual(cached.cond.notify_count, 3) def test_decorator_info(self): cache = self.cache(2) cached = Cached(cache) wrapper = cached.get_info maxsize = cache.maxsize if isinstance(cache, Cache) else None self.assertEqual(wrapper.cache_info(), (0, 0, maxsize, 0)) self.assertEqual(wrapper(0), 0) self.assertEqual(wrapper.cache_info(), (0, 1, maxsize, 1)) self.assertEqual(wrapper(1), 1) self.assertEqual(wrapper.cache_info(), (0, 2, maxsize, 2)) self.assertEqual(wrapper(0), 0) self.assertEqual(wrapper.cache_info(), (1, 2, maxsize, 2)) wrapper.cache_clear() self.assertEqual(len(cache), 0) self.assertEqual(wrapper.cache_info(), (0, 0, maxsize, 0)) self.assertEqual(wrapper(0), 2) self.assertEqual(wrapper.cache_info(), (0, 1, maxsize, 1)) self.assertEqual(wrapper(0), 2) self.assertEqual(wrapper.cache_info(), (1, 1, maxsize, 1)) # assert hits/misses are counted per instance cached = Cached(self.cache(2)) self.assertEqual(cached.get_info(0), 0) self.assertEqual(cached.get_info.cache_info(), (0, 1, maxsize, 1)) def test_decorator_lock_info(self): cache = self.cache(2) cached = Cached(cache) wrapper = cached.get_lock_info maxsize = cache.maxsize if isinstance(cache, Cache) else None self.assertEqual(wrapper.cache_info(), (0, 0, maxsize, 0)) self.assertEqual(cached.lock.count, 1) self.assertEqual(wrapper(0), 0) self.assertEqual(cached.lock.count, 3) self.assertEqual(wrapper.cache_info(), (0, 1, maxsize, 1)) self.assertEqual(cached.lock.count, 4) self.assertEqual(wrapper(1), 1) self.assertEqual(cached.lock.count, 6) self.assertEqual(wrapper.cache_info(), (0, 2, maxsize, 2)) self.assertEqual(cached.lock.count, 7) self.assertEqual(wrapper(0), 0) self.assertEqual(cached.lock.count, 8) self.assertEqual(wrapper.cache_info(), (1, 2, maxsize, 2)) self.assertEqual(cached.lock.count, 9) wrapper.cache_clear() self.assertEqual(cached.lock.count, 10) self.assertEqual(len(cache), 0) self.assertEqual(cached.lock.count, 10) self.assertEqual(wrapper.cache_info(), (0, 0, maxsize, 0)) self.assertEqual(cached.lock.count, 11) self.assertEqual(wrapper(0), 2) self.assertEqual(cached.lock.count, 13) self.assertEqual(wrapper.cache_info(), (0, 1, maxsize, 1)) self.assertEqual(cached.lock.count, 14) self.assertEqual(wrapper(0), 2) self.assertEqual(cached.lock.count, 15) self.assertEqual(wrapper.cache_info(), (1, 1, maxsize, 1)) self.assertEqual(cached.lock.count, 16) # assert hits/misses are counted per instance cached = Cached(self.cache(2)) self.assertEqual(cached.get_lock_info(0), 0) self.assertEqual(cached.get_lock_info.cache_info(), (0, 1, maxsize, 1)) def test_decorator_cond_info(self): cache = self.cache(2) cached = Cached(cache) wrapper = cached.get_cond_info maxsize = cache.maxsize if isinstance(cache, Cache) else None self.assertEqual(wrapper.cache_info(), (0, 0, maxsize, 0)) self.assertEqual(cached.cond.count, 1) self.assertEqual(wrapper(0), 0) self.assertEqual(cached.cond.count, 4) self.assertEqual(cached.cond.wait_count, 1) self.assertEqual(cached.cond.notify_count, 1) self.assertEqual(wrapper.cache_info(), (0, 1, maxsize, 1)) self.assertEqual(cached.cond.count, 5) self.assertEqual(wrapper(1), 1) self.assertEqual(cached.cond.count, 8) self.assertEqual(cached.cond.wait_count, 2) self.assertEqual(cached.cond.notify_count, 2) self.assertEqual(wrapper.cache_info(), (0, 2, maxsize, 2)) self.assertEqual(cached.cond.count, 9) self.assertEqual(wrapper(0), 0) self.assertEqual(cached.cond.count, 10) self.assertEqual(cached.cond.wait_count, 3) self.assertEqual(cached.cond.notify_count, 2) self.assertEqual(wrapper.cache_info(), (1, 2, maxsize, 2)) self.assertEqual(cached.cond.count, 11) wrapper.cache_clear() self.assertEqual(cached.cond.count, 12) self.assertEqual(len(cache), 0) self.assertEqual(cached.cond.count, 12) self.assertEqual(wrapper.cache_info(), (0, 0, maxsize, 0)) self.assertEqual(cached.cond.count, 13) self.assertEqual(wrapper(0), 2) self.assertEqual(cached.cond.count, 16) self.assertEqual(cached.cond.wait_count, 4) self.assertEqual(cached.cond.notify_count, 3) self.assertEqual(wrapper.cache_info(), (0, 1, maxsize, 1)) self.assertEqual(cached.cond.count, 17) self.assertEqual(wrapper(0), 2) self.assertEqual(cached.cond.count, 18) self.assertEqual(cached.cond.wait_count, 5) self.assertEqual(cached.cond.notify_count, 3) self.assertEqual(wrapper.cache_info(), (1, 1, maxsize, 1)) self.assertEqual(cached.cond.count, 19) # assert hits/misses are counted per instance cached = Cached(self.cache(2)) self.assertEqual(cached.get_cond_info(0), 0) self.assertEqual(cached.get_cond_info.cache_info(), (0, 1, maxsize, 1)) def test_decorator_lock_cond_info(self): cache = self.cache(2) cached = Cached(cache) wrapper = cached.get_lock_cond_info maxsize = cache.maxsize if isinstance(cache, Cache) else None self.assertEqual(wrapper.cache_info(), (0, 0, maxsize, 0)) self.assertEqual(cached.lock.count, 1) self.assertEqual(wrapper(0), 0) self.assertEqual(cached.lock.count, 4) self.assertEqual(cached.cond.wait_count, 1) self.assertEqual(cached.cond.notify_count, 1) self.assertEqual(wrapper.cache_info(), (0, 1, maxsize, 1)) self.assertEqual(cached.lock.count, 5) self.assertEqual(wrapper(1), 1) self.assertEqual(cached.lock.count, 8) self.assertEqual(cached.cond.wait_count, 2) self.assertEqual(cached.cond.notify_count, 2) self.assertEqual(wrapper.cache_info(), (0, 2, maxsize, 2)) self.assertEqual(cached.lock.count, 9) self.assertEqual(wrapper(0), 0) self.assertEqual(cached.lock.count, 10) self.assertEqual(cached.cond.wait_count, 3) self.assertEqual(cached.cond.notify_count, 2) self.assertEqual(wrapper.cache_info(), (1, 2, maxsize, 2)) self.assertEqual(cached.lock.count, 11) wrapper.cache_clear() self.assertEqual(cached.lock.count, 12) self.assertEqual(len(cache), 0) self.assertEqual(cached.lock.count, 12) self.assertEqual(wrapper.cache_info(), (0, 0, maxsize, 0)) self.assertEqual(cached.lock.count, 13) self.assertEqual(wrapper(0), 2) self.assertEqual(cached.lock.count, 16) self.assertEqual(cached.cond.wait_count, 4) self.assertEqual(cached.cond.notify_count, 3) self.assertEqual(wrapper.cache_info(), (0, 1, maxsize, 1)) self.assertEqual(cached.lock.count, 17) self.assertEqual(wrapper(0), 2) self.assertEqual(cached.lock.count, 18) self.assertEqual(cached.cond.wait_count, 5) self.assertEqual(cached.cond.notify_count, 3) self.assertEqual(wrapper.cache_info(), (1, 1, maxsize, 1)) self.assertEqual(cached.lock.count, 19) # assert hits/misses are counted per instance cached = Cached(self.cache(2)) self.assertEqual(cached.get_lock_cond_info(0), 0) self.assertEqual(cached.get_lock_cond_info.cache_info(), (0, 1, maxsize, 1)) def test_decorator_cond_error(self): cached = Cached(self.cache(2)) with self.assertRaises(ValueError): cached.get_cond_error(0) self.assertEqual(cached.cond.count, 2) self.assertEqual(cached.cond.wait_count, 1) self.assertEqual(cached.cond.notify_count, 1) # verify pending set is cleaned up, otherwise this might deadlock with self.assertRaises(ValueError): cached.get_cond_error(0) self.assertEqual(cached.cond.count, 4) self.assertEqual(cached.cond.wait_count, 2) self.assertEqual(cached.cond.notify_count, 2) def test_decorator_wrapped(self): cache = self.cache(2) cached = Cached(cache) self.assertEqual(len(cache), 0) self.assertEqual(cached.get.__wrapped__(cached, 0), 0) self.assertEqual(cached.get.__name__, "get") self.assertEqual(cached.get.__doc__.strip(), "docstring") self.assertEqual(len(cache), 0) self.assertEqual(cached.get(0), 1) self.assertEqual(len(cache), 1) self.assertEqual(cached.get(0), 1) self.assertEqual(len(cache), 1) def test_decorator_attributes(self): cache = self.cache(2) cached = Cached(cache) self.assertIs(cached.get.cache, cache) self.assertIs(cached.get.cache_key, keys.methodkey) self.assertIs(cached.get.cache_lock, None) self.assertIs(cached.get.cache_condition, None) self.assertIs(cached.get_lock.cache, cache) self.assertIs(cached.get_lock.cache_key, keys.methodkey) self.assertIs(cached.get_lock.cache_lock, cached.lock) self.assertIs(cached.get_lock.cache_condition, None) self.assertIs(cached.get_cond.cache, cache) self.assertIs(cached.get_cond.cache_key, keys.methodkey) self.assertIs(cached.get_cond.cache_lock, cached.cond) self.assertIs(cached.get_cond.cache_condition, cached.cond) self.assertIs(cached.get_lock_cond_info.cache, cache) self.assertIs(cached.get_lock_cond_info.cache_key, keys.methodkey) self.assertIs(cached.get_lock_cond_info.cache_lock, cached.lock) self.assertIs(cached.get_lock_cond_info.cache_condition, cached.cond) def test_decorator_clear(self): cache = self.cache(2) cached = Cached(cache) self.assertEqual(cached.get(0), 0) self.assertEqual(len(cache), 1) cached.get.cache_clear() self.assertEqual(len(cache), 0) def test_decorator_lock_clear(self): cache = self.cache(2) cached = Cached(cache) self.assertEqual(cached.get_lock(0), 0) self.assertEqual(len(cache), 1) self.assertEqual(cached.lock.count, 2) cached.get_lock.cache_clear() self.assertEqual(len(cache), 0) self.assertEqual(cached.lock.count, 3) def test_decorator_cond_clear(self): cache = self.cache(2) cached = Cached(cache) self.assertEqual(cached.get_cond(0), 0) self.assertEqual(len(cache), 1) self.assertEqual(cached.cond.count, 3) cached.get_cond.cache_clear() self.assertEqual(len(cache), 0) self.assertEqual(cached.cond.count, 4) def test_decorator_slots(self): class Slots: __slots__ = ("cache",) def __init__(self, cache): self.cache = cache @cachedmethod(lambda self: self.cache) def get(self, value): return value @cachedmethod(lambda self: self.cache, info=True) def get_info(self, value): return value cache = self.cache(2) cached = Slots(cache) with self.assertRaises(TypeError): cached.get_info(42) with warnings.catch_warnings(record=True) as w: warnings.simplefilter("always") self.assertEqual(cached.get(0), 0) self.assertEqual(cached.get(0), 0) self.assertEqual(len(cache), 1) self.assertEqual(len(w), 2) self.assertIs(w[0].category, DeprecationWarning) self.assertIs(w[1].category, DeprecationWarning) def test_decorator_immutable_dict(self): class ReadOnlyDict(dict): def __setitem__(self, _key, _value): raise TypeError("Modification not supported") def setdefault(self, _key, _value): raise TypeError("Modification not supported") class Immutable: def __init__(self, cache): self.cache = cache self.__dict__ = ReadOnlyDict(self.__dict__) @cachedmethod(lambda self: self.cache) def get(self, value): return value @cachedmethod(lambda self: self.cache, info=True) def get_info(self, value): return value cache = self.cache(2) cached = Immutable(cache) with self.assertRaises(TypeError): cached.get_info(42) with warnings.catch_warnings(record=True) as w: warnings.simplefilter("always") self.assertEqual(cached.get(0), 0) self.assertEqual(cached.get(0), 0) self.assertEqual(len(cache), 1) self.assertEqual(len(w), 2) self.assertIs(w[0].category, DeprecationWarning) self.assertIs(w[1].category, DeprecationWarning) def test_decorator_different_names(self): # RuntimeError for Python < 3.12, TypeError otherwise with self.assertRaisesRegex(Exception, "bar"): class Cached: @cachedmethod(lambda _: self.cache(2)) def foo(_self): pass bar = foo def test_decorator_no_set_name(self): class Cached: def foo(_self): pass # __set_name__ currently only asserted with info, since this # may also occur with deprecated @classmethod use Cached.bar = cachedmethod(lambda _: self.cache(2), info=True)(Cached.foo) cached = Cached() with self.assertRaises(TypeError): cached.bar() class CacheMethodTest(unittest.TestCase, MethodDecoratorTestMixin): def cache(self, minsize, **kwargs): return Cache(maxsize=minsize, **kwargs) def test_nospace(self): cached = Cached(self.cache(0)) self.assertEqual(cached.get(0), 0) self.assertEqual(cached.get(1), 1) self.assertEqual(cached.get(1), 2) self.assertEqual(cached.get(1.0), 3) self.assertEqual(cached.get(1.0), 4) self.assertEqual(cached.get_typed(0), 5) self.assertEqual(cached.get_typed(1), 6) self.assertEqual(cached.get_typed(1.0), 7) self.assertEqual(cached.get_info(0), 8) self.assertEqual(cached.get_info(1), 9) self.assertEqual(cached.get_info(1.0), 10) def test_lock_nospace(self): cached = Cached(self.cache(0)) self.assertEqual(cached.get_lock(0), 0) self.assertEqual(cached.lock.count, 2) self.assertEqual(cached.get_lock(1), 1) self.assertEqual(cached.lock.count, 4) self.assertEqual(cached.get_lock(1), 2) self.assertEqual(cached.lock.count, 6) self.assertEqual(cached.get_lock(1.0), 3) self.assertEqual(cached.lock.count, 8) self.assertEqual(cached.get_lock(1.0), 4) self.assertEqual(cached.lock.count, 10) self.assertEqual(cached.get_lock_info(0), 5) self.assertEqual(cached.lock.count, 12) self.assertEqual(cached.get_lock_info(1), 6) self.assertEqual(cached.lock.count, 14) self.assertEqual(cached.get_lock_info(1.0), 7) self.assertEqual(cached.lock.count, 16) def test_cond_nospace(self): cached = Cached(self.cache(0)) self.assertEqual(cached.get_cond(0), 0) self.assertEqual(cached.cond.count, 3) self.assertEqual(cached.cond.wait_count, 1) self.assertEqual(cached.cond.notify_count, 1) self.assertEqual(cached.get_cond(1), 1) self.assertEqual(cached.cond.count, 6) self.assertEqual(cached.cond.wait_count, 2) self.assertEqual(cached.cond.notify_count, 2) self.assertEqual(cached.get_cond(1), 2) self.assertEqual(cached.cond.count, 9) self.assertEqual(cached.cond.wait_count, 3) self.assertEqual(cached.cond.notify_count, 3) self.assertEqual(cached.get_cond(1.0), 3) self.assertEqual(cached.cond.count, 12) self.assertEqual(cached.cond.wait_count, 4) self.assertEqual(cached.cond.notify_count, 4) self.assertEqual(cached.get_cond(1.0), 4) self.assertEqual(cached.cond.count, 15) self.assertEqual(cached.cond.wait_count, 5) self.assertEqual(cached.cond.notify_count, 5) self.assertEqual(cached.get_cond_info(0), 5) self.assertEqual(cached.cond.count, 18) self.assertEqual(cached.cond.wait_count, 6) self.assertEqual(cached.cond.notify_count, 6) self.assertEqual(cached.get_cond_info(1), 6) self.assertEqual(cached.cond.count, 21) self.assertEqual(cached.cond.wait_count, 7) self.assertEqual(cached.cond.notify_count, 7) self.assertEqual(cached.get_cond(1.0), 7) self.assertEqual(cached.cond.count, 24) self.assertEqual(cached.cond.wait_count, 8) self.assertEqual(cached.cond.notify_count, 8) def test_shared_cache(self): cache = self.cache(2) cached1 = Cached(cache) cached2 = Cached(cache) self.assertEqual(cached1.get_info.cache_info(), (0, 0, 2, 0)) self.assertEqual(cached2.get_info.cache_info(), (0, 0, 2, 0)) self.assertEqual(cached1.get_info(0), 0) self.assertEqual(cached1.get_info.cache_info(), (0, 1, 2, 1)) self.assertEqual(cached2.get_info.cache_info(), (0, 0, 2, 1)) self.assertEqual(cached2.get_info(0), 0) self.assertEqual(cached1.get_info.cache_info(), (0, 1, 2, 1)) self.assertEqual(cached2.get_info.cache_info(), (1, 0, 2, 1)) def test_value_too_large(self): cache = self.cache(1, getsizeof=lambda x: x) cached = Cached(cache) self.assertEqual(cached.get(0), 0) self.assertIn(0, cache.values()) self.assertEqual(cached.get(1), 1) self.assertIn(1, cache.values()) self.assertEqual(cached.get(2), 2) self.assertNotIn(2, cache.values()) class DictMethodTest(unittest.TestCase, MethodDecoratorTestMixin): def cache(self, _minsize): return dict() class WeakRefMethodTest(unittest.TestCase): def test_weakref(self): import fractions import gc import weakref # in Python 3.9, `int` does not support weak references even # when subclassed, but Fraction apparently does... class Int(fractions.Fraction): def __add__(self, other): return Int(fractions.Fraction.__add__(self, other)) cache = weakref.WeakValueDictionary() cached = Cached(cache, count=Int(0)) self.assertEqual(cached.get(0), 0) gc.collect() self.assertEqual(cached.get(0), 1) ref = cached.get(1) self.assertEqual(ref, 2) self.assertEqual(cached.get(1), 2) self.assertEqual(cached.get(1.0), 2) ref = cached.get_typed(2) self.assertEqual(ref, 3) self.assertEqual(cached.get_typed(1), 4) self.assertEqual(cached.get_typed(1.0), 5) cached.cache.clear() self.assertEqual(cached.get(1), 6) class NoneMethodTest(unittest.TestCase): def cache(self, _minsize): return None def test_none_info(self): cached = Cached(self.cache(1)) wrapper = cached.get_info with self.assertRaises(TypeError): wrapper.cache_info()