import unittest import warnings import cachetools import cachetools.keys from . import CountedCondition, CountedLock, SimplifiedTestCaseProtocol class DecoratorTestMixin(SimplifiedTestCaseProtocol): def cache(self, minsize): raise NotImplementedError def func(self, *args, **kwargs): if hasattr(self, "count"): self.count += 1 else: self.count = 0 return self.count def error_func(self, *args, **kwargs): raise ValueError("test error") def test_decorator(self): cache = self.cache(2) wrapper = cachetools.cached(cache)(self.func) self.assertEqual(len(cache), 0) self.assertEqual(wrapper(0), 0) self.assertEqual(len(cache), 1) self.assertIn(cachetools.keys.hashkey(0), cache) self.assertNotIn(cachetools.keys.hashkey(1), cache) self.assertNotIn(cachetools.keys.hashkey(1.0), cache) self.assertEqual(wrapper(1), 1) self.assertEqual(len(cache), 2) self.assertIn(cachetools.keys.hashkey(0), cache) self.assertIn(cachetools.keys.hashkey(1), cache) self.assertIn(cachetools.keys.hashkey(1.0), cache) self.assertEqual(wrapper(1), 1) self.assertEqual(len(cache), 2) self.assertEqual(wrapper(1.0), 1) self.assertEqual(len(cache), 2) self.assertEqual(wrapper(1.0), 1) self.assertEqual(len(cache), 2) def test_decorator_typed(self): cache = self.cache(3) key = cachetools.keys.typedkey wrapper = cachetools.cached(cache, key=key)(self.func) self.assertEqual(len(cache), 0) self.assertEqual(wrapper(0), 0) self.assertEqual(len(cache), 1) self.assertIn(cachetools.keys.typedkey(0), cache) self.assertNotIn(cachetools.keys.typedkey(1), cache) self.assertNotIn(cachetools.keys.typedkey(1.0), cache) self.assertEqual(wrapper(1), 1) self.assertEqual(len(cache), 2) self.assertIn(cachetools.keys.typedkey(0), cache) self.assertIn(cachetools.keys.typedkey(1), cache) self.assertNotIn(cachetools.keys.typedkey(1.0), cache) self.assertEqual(wrapper(1), 1) self.assertEqual(len(cache), 2) self.assertEqual(wrapper(1.0), 2) self.assertEqual(len(cache), 3) self.assertIn(cachetools.keys.typedkey(0), cache) self.assertIn(cachetools.keys.typedkey(1), cache) self.assertIn(cachetools.keys.typedkey(1.0), cache) self.assertEqual(wrapper(1.0), 2) self.assertEqual(len(cache), 3) def test_decorator_lock(self): cache = self.cache(2) lock = CountedLock() wrapper = cachetools.cached(cache, lock=lock)(self.func) self.assertEqual(len(cache), 0) self.assertEqual(wrapper(0), 0) self.assertEqual(lock.count, 2) self.assertEqual(wrapper(1), 1) self.assertEqual(lock.count, 4) self.assertEqual(wrapper(1), 1) self.assertEqual(lock.count, 5) def test_decorator_cond(self): cache = self.cache(2) lock = cond = CountedCondition() wrapper = cachetools.cached(cache, condition=cond)(self.func) self.assertEqual(len(cache), 0) self.assertEqual(wrapper(0), 0) self.assertEqual(lock.count, 3) self.assertEqual(cond.wait_count, 1) self.assertEqual(cond.notify_count, 1) self.assertEqual(wrapper(1), 1) self.assertEqual(lock.count, 6) self.assertEqual(cond.wait_count, 2) self.assertEqual(cond.notify_count, 2) self.assertEqual(wrapper(1), 1) self.assertEqual(lock.count, 7) self.assertEqual(cond.wait_count, 3) self.assertEqual(cond.notify_count, 2) def test_decorator_lock_cond(self): cache = self.cache(2) lock = CountedLock() cond = CountedCondition() wrapper = cachetools.cached(cache, lock=lock, condition=cond)(self.func) self.assertEqual(len(cache), 0) self.assertEqual(wrapper(0), 0) self.assertEqual(lock.count, 3) self.assertEqual(cond.wait_count, 1) self.assertEqual(cond.notify_count, 1) self.assertEqual(wrapper(1), 1) self.assertEqual(lock.count, 6) self.assertEqual(cond.wait_count, 2) self.assertEqual(cond.notify_count, 2) self.assertEqual(wrapper(1), 1) self.assertEqual(lock.count, 7) self.assertEqual(cond.wait_count, 3) self.assertEqual(cond.notify_count, 2) def test_decorator_cond_error(self): cache = self.cache(2) lock = cond = CountedCondition() wrapper = cachetools.cached(cache, condition=cond)(self.error_func) with self.assertRaises(ValueError): wrapper(0) self.assertEqual(len(cache), 0) self.assertEqual(lock.count, 2) self.assertEqual(cond.wait_count, 1) self.assertEqual(cond.notify_count, 1) # verify pending set is cleaned up, otherwise this might deadlock with self.assertRaises(ValueError): wrapper(0) self.assertEqual(len(cache), 0) self.assertEqual(lock.count, 4) self.assertEqual(cond.wait_count, 2) self.assertEqual(cond.notify_count, 2) def test_decorator_wrapped(self): cache = self.cache(2) wrapper = cachetools.cached(cache)(self.func) self.assertEqual(wrapper.__wrapped__, self.func) self.assertEqual(wrapper.__name__, "func") self.assertEqual(wrapper.__qualname__, "DecoratorTestMixin.func") self.assertEqual(len(cache), 0) self.assertEqual(wrapper.__wrapped__(0), 0) self.assertEqual(len(cache), 0) self.assertEqual(wrapper(0), 1) self.assertEqual(len(cache), 1) self.assertEqual(wrapper(0), 1) self.assertEqual(len(cache), 1) def test_decorator_attributes(self): cache = self.cache(2) wrapper = cachetools.cached(cache)(self.func) self.assertIs(wrapper.cache, cache) self.assertIs(wrapper.cache_key, cachetools.keys.hashkey) self.assertIs(wrapper.cache_lock, None) self.assertIs(wrapper.cache_condition, None) def test_decorator_attributes_lock(self): cache = self.cache(2) lock = CountedLock() wrapper = cachetools.cached(cache, lock=lock)(self.func) self.assertIs(wrapper.cache, cache) self.assertIs(wrapper.cache_key, cachetools.keys.hashkey) self.assertIs(wrapper.cache_lock, lock) self.assertIs(wrapper.cache_condition, None) def test_decorator_attributes_cond(self): cache = self.cache(2) lock = cond = CountedCondition() wrapper = cachetools.cached(cache, condition=cond)(self.func) self.assertIs(wrapper.cache, cache) self.assertIs(wrapper.cache_key, cachetools.keys.hashkey) self.assertIs(wrapper.cache_lock, lock) self.assertIs(wrapper.cache_condition, cond) def test_decorator_clear(self): cache = self.cache(2) wrapper = cachetools.cached(cache)(self.func) self.assertEqual(wrapper(0), 0) self.assertEqual(len(cache), 1) wrapper.cache_clear() self.assertEqual(len(cache), 0) def test_decorator_clear_lock(self): cache = self.cache(2) lock = CountedLock() wrapper = cachetools.cached(cache, lock=lock)(self.func) self.assertEqual(wrapper(0), 0) self.assertEqual(len(cache), 1) self.assertEqual(lock.count, 2) wrapper.cache_clear() self.assertEqual(len(cache), 0) self.assertEqual(lock.count, 3) def test_decorator_clear_cond(self): cache = self.cache(2) lock = cond = CountedCondition() wrapper = cachetools.cached(cache, condition=cond)(self.func) self.assertEqual(wrapper(0), 0) self.assertEqual(len(cache), 1) self.assertEqual(lock.count, 3) wrapper.cache_clear() self.assertEqual(len(cache), 0) self.assertEqual(lock.count, 4) class CacheWrapperTest(unittest.TestCase, DecoratorTestMixin): def cache(self, minsize): return cachetools.Cache(maxsize=minsize) def test_decorator_info(self): cache = self.cache(2) wrapper = cachetools.cached(cache, info=True)(self.func) self.assertEqual(wrapper.cache_info(), (0, 0, 2, 0)) self.assertEqual(wrapper(0), 0) self.assertEqual(wrapper.cache_info(), (0, 1, 2, 1)) self.assertEqual(wrapper(1), 1) self.assertEqual(wrapper.cache_info(), (0, 2, 2, 2)) self.assertEqual(wrapper(0), 0) self.assertEqual(wrapper.cache_info(), (1, 2, 2, 2)) wrapper.cache_clear() self.assertEqual(len(cache), 0) self.assertEqual(wrapper.cache_info(), (0, 0, 2, 0)) def test_decorator_lock_info(self): cache = self.cache(2) lock = CountedLock() wrapper = cachetools.cached(cache, lock=lock, info=True)(self.func) self.assertEqual(wrapper.cache_info(), (0, 0, 2, 0)) self.assertEqual(lock.count, 1) self.assertEqual(wrapper(0), 0) self.assertEqual(lock.count, 3) self.assertEqual(wrapper.cache_info(), (0, 1, 2, 1)) self.assertEqual(lock.count, 4) self.assertEqual(wrapper(1), 1) self.assertEqual(lock.count, 6) self.assertEqual(wrapper.cache_info(), (0, 2, 2, 2)) self.assertEqual(lock.count, 7) self.assertEqual(wrapper(0), 0) self.assertEqual(lock.count, 8) self.assertEqual(wrapper.cache_info(), (1, 2, 2, 2)) self.assertEqual(lock.count, 9) wrapper.cache_clear() self.assertEqual(lock.count, 10) self.assertEqual(len(cache), 0) self.assertEqual(wrapper.cache_info(), (0, 0, 2, 0)) self.assertEqual(lock.count, 11) def test_decorator_cond_info(self): cache = self.cache(2) lock = cond = CountedCondition() wrapper = cachetools.cached(cache, condition=cond, info=True)(self.func) self.assertEqual(wrapper.cache_info(), (0, 0, 2, 0)) self.assertEqual(lock.count, 1) self.assertEqual(wrapper(0), 0) self.assertEqual(lock.count, 4) self.assertEqual(wrapper.cache_info(), (0, 1, 2, 1)) self.assertEqual(lock.count, 5) self.assertEqual(wrapper(1), 1) self.assertEqual(lock.count, 8) self.assertEqual(wrapper.cache_info(), (0, 2, 2, 2)) self.assertEqual(lock.count, 9) self.assertEqual(wrapper(0), 0) self.assertEqual(lock.count, 10) self.assertEqual(wrapper.cache_info(), (1, 2, 2, 2)) self.assertEqual(lock.count, 11) wrapper.cache_clear() self.assertEqual(lock.count, 12) self.assertEqual(len(cache), 0) self.assertEqual(wrapper.cache_info(), (0, 0, 2, 0)) self.assertEqual(lock.count, 13) def test_decorator_lock_cond_info(self): cache = self.cache(2) lock = CountedLock() cond = CountedCondition() wrapper = cachetools.cached(cache, lock=lock, condition=cond, info=True)( self.func ) self.assertEqual(wrapper.cache_info(), (0, 0, 2, 0)) self.assertEqual(lock.count, 1) self.assertEqual(wrapper(0), 0) self.assertEqual(lock.count, 4) self.assertEqual(wrapper.cache_info(), (0, 1, 2, 1)) self.assertEqual(lock.count, 5) self.assertEqual(wrapper(1), 1) self.assertEqual(lock.count, 8) self.assertEqual(wrapper.cache_info(), (0, 2, 2, 2)) self.assertEqual(lock.count, 9) self.assertEqual(wrapper(0), 0) self.assertEqual(lock.count, 10) self.assertEqual(wrapper.cache_info(), (1, 2, 2, 2)) self.assertEqual(lock.count, 11) wrapper.cache_clear() self.assertEqual(lock.count, 12) self.assertEqual(len(cache), 0) self.assertEqual(wrapper.cache_info(), (0, 0, 2, 0)) self.assertEqual(lock.count, 13) def test_stacked_decorators(self): inner_cache = self.cache(2) outer_cache = self.cache(2) inner_wrapper = cachetools.cached(inner_cache, info=True)(self.func) outer_wrapper = cachetools.cached(outer_cache, info=True)(inner_wrapper) self.assertIs(outer_wrapper.__wrapped__, inner_wrapper) self.assertIs(outer_wrapper.cache, outer_cache) self.assertEqual(outer_wrapper(0), 0) self.assertEqual(outer_wrapper(0), 0) self.assertEqual(outer_wrapper.cache_info(), (1, 1, 2, 1)) self.assertEqual(inner_wrapper.cache_info(), (0, 1, 2, 1)) outer_wrapper.cache_clear() self.assertEqual(len(outer_cache), 0) self.assertEqual(len(inner_cache), 1) def test_zero_size_cache_decorator(self): cache = self.cache(0) wrapper = cachetools.cached(cache)(self.func) self.assertEqual(len(cache), 0) self.assertEqual(wrapper(0), 0) self.assertEqual(len(cache), 0) def test_zero_size_cache_decorator_lock(self): cache = self.cache(0) lock = CountedLock() wrapper = cachetools.cached(cache, lock=lock)(self.func) self.assertEqual(len(cache), 0) self.assertEqual(wrapper(0), 0) self.assertEqual(len(cache), 0) self.assertEqual(lock.count, 2) def test_zero_size_cache_decorator_cond(self): cache = self.cache(0) lock = cond = CountedCondition() wrapper = cachetools.cached(cache, condition=cond)(self.func) self.assertEqual(len(cache), 0) self.assertEqual(wrapper(0), 0) self.assertEqual(len(cache), 0) self.assertEqual(lock.count, 3) def test_zero_size_cache_decorator_info(self): cache = self.cache(0) wrapper = cachetools.cached(cache, info=True)(self.func) self.assertEqual(wrapper.cache_info(), (0, 0, 0, 0)) self.assertEqual(wrapper(0), 0) self.assertEqual(wrapper.cache_info(), (0, 1, 0, 0)) def test_zero_size_cache_decorator_lock_info(self): cache = self.cache(0) lock = CountedLock() wrapper = cachetools.cached(cache, lock=lock, info=True)(self.func) self.assertEqual(len(cache), 0) self.assertEqual(wrapper.cache_info(), (0, 0, 0, 0)) self.assertEqual(lock.count, 1) self.assertEqual(wrapper(0), 0) self.assertEqual(len(cache), 0) self.assertEqual(lock.count, 3) self.assertEqual(wrapper.cache_info(), (0, 1, 0, 0)) self.assertEqual(lock.count, 4) def test_zero_size_cache_decorator_cond_info(self): cache = self.cache(0) lock = cond = CountedCondition() wrapper = cachetools.cached(cache, condition=cond, info=True)(self.func) self.assertEqual(len(cache), 0) self.assertEqual(wrapper.cache_info(), (0, 0, 0, 0)) self.assertEqual(lock.count, 1) self.assertEqual(wrapper(0), 0) self.assertEqual(len(cache), 0) self.assertEqual(lock.count, 4) self.assertEqual(wrapper.cache_info(), (0, 1, 0, 0)) self.assertEqual(lock.count, 5) def test_decorator_method(self): """Though it is not encouraged, like functools.lru_cache, the @cached decorator may be used on methods, too. """ cache = self.cache(2) class Cached: @cachetools.cached(cache, key=cachetools.keys.methodkey) def get(self, value: int) -> int: return value @cachetools.cached(self.cache(2), key=cachetools.keys.methodkey, info=True) def get_info(self, value): return value @staticmethod @cachetools.cached(self.cache(2)) def get_static(value: int) -> int: return value cached = Cached() self.assertEqual(cached.get(1), 1) self.assertEqual(Cached.get(cached, 1), 1) self.assertEqual(len(cache), 1) cached.get.cache_clear() self.assertEqual(len(cache), 0) self.assertEqual(cached.get_info(1), 1) self.assertEqual(cached.get_info(1), 1) self.assertEqual(Cached.get_info(cached, 1), 1) self.assertEqual(cached.get_info.cache_info(), (2, 1, 2, 1)) self.assertEqual(Cached.get_info.cache_info(), (2, 1, 2, 1)) cached.get_info.cache_clear() self.assertEqual(cached.get_info.cache_info(), (0, 0, 2, 0)) self.assertEqual(Cached.get_info.cache_info(), (0, 0, 2, 0)) self.assertEqual(cached.get_static(1), 1) self.assertEqual(Cached.get_static(1), 1) class DictWrapperTest(unittest.TestCase, DecoratorTestMixin): def cache(self, minsize): return {} def test_decorator_info(self): cache = self.cache(2) wrapper = cachetools.cached(cache, info=True)(self.func) self.assertEqual(wrapper.cache_info(), (0, 0, None, 0)) self.assertEqual(wrapper(0), 0) self.assertEqual(wrapper.cache_info(), (0, 1, None, 1)) self.assertEqual(wrapper(1), 1) self.assertEqual(wrapper.cache_info(), (0, 2, None, 2)) self.assertEqual(wrapper(0), 0) self.assertEqual(wrapper.cache_info(), (1, 2, None, 2)) wrapper.cache_clear() self.assertEqual(len(cache), 0) self.assertEqual(wrapper.cache_info(), (0, 0, None, 0)) class NoneWrapperTest(unittest.TestCase): def func(self, *args, **kwargs): return args + tuple(kwargs.items()) def test_decorator(self): with warnings.catch_warnings(record=True) as w: wrapper = cachetools.cached(None)(self.func) self.assertIs(w[0].category, DeprecationWarning) self.assertEqual(wrapper(0), (0,)) self.assertEqual(wrapper(1), (1,)) self.assertEqual(wrapper(1, foo="bar"), (1, ("foo", "bar"))) def test_decorator_attributes(self): with warnings.catch_warnings(record=True) as w: wrapper = cachetools.cached(None)(self.func) self.assertIs(w[0].category, DeprecationWarning) self.assertIs(wrapper.cache, None) self.assertIs(wrapper.cache_key, cachetools.keys.hashkey) self.assertIs(wrapper.cache_lock, None) def test_decorator_clear(self): with warnings.catch_warnings(record=True) as w: wrapper = cachetools.cached(None)(self.func) self.assertIs(w[0].category, DeprecationWarning) wrapper.cache_clear() # no-op def test_decorator_info(self): with warnings.catch_warnings(record=True) as w: wrapper = cachetools.cached(None, info=True)(self.func) self.assertIs(w[0].category, DeprecationWarning) self.assertEqual(wrapper.cache_info(), (0, 0, 0, 0)) self.assertEqual(wrapper(0), (0,)) self.assertEqual(wrapper.cache_info(), (0, 1, 0, 0)) self.assertEqual(wrapper(1), (1,)) self.assertEqual(wrapper.cache_info(), (0, 2, 0, 0)) wrapper.cache_clear() self.assertEqual(wrapper.cache_info(), (0, 0, 0, 0))