import pytest from caching.BaseCache import BaseCache, MAX_INITIALIZED_KEY from caching.Cache import Cache from caching.DictionaryCache import DictionaryCache from caching.IncCache import IncCache from caching.ListCache import ListCache from caching.ListIfNeededCache import ListIfNeededCache from caching.SetCache import SetCache from common.global_symbols import NotFound, Removed from tests.caching import FakeSdp class TestCache: def test_i_can_configure(self): cache = Cache() cache.configure(max_size=256, default="default_delegate", extend_exists="extend_exists_delegate", alt_sdp_get="alt_sdp_delegate", sdp=FakeSdp()) # Caution, in this test, I initialize default, extend_exists and alt_get_delegate with string # to simplify the test, but in real usage, they are lambda # default = lambda sdp, key: sdp.get(cache_name, key) or lambda key: func(key) # extend_exists = lambda sdp, key: sdp.exists(cache_name, key) or lambda key: func(key) # alt_sdp_get = lambda sdp, key: sdp.alt_get(cache_name, key) assert cache._max_size == 256 assert cache._default == "default_delegate" assert cache._extend_exists == "extend_exists_delegate" assert cache._alt_sdp_get == "alt_sdp_delegate" assert cache._sdp is not None def test_i_can_auto_configure(self): sdp = FakeSdp(get_value=lambda cache_name, key: key + 1 if cache_name == "cache_name" else NotFound, extend_exists=lambda cache_name, key: True if cache_name == "cache_name" else False, get_alt_value=lambda cache_name, key: key + 2 if cache_name == "cache_name" else NotFound) cache = Cache(sdp=sdp).auto_configure("cache_name") assert cache._default(cache._sdp, 10) == 11 assert cache._extend_exists(cache._sdp, 10) is True assert cache._alt_sdp_get(cache._sdp, 10) == 12 cache = Cache(sdp=sdp).auto_configure("another_cache") assert cache._default(cache._sdp, 10) == NotFound assert cache._extend_exists(cache._sdp, 10) is False assert cache._alt_sdp_get(cache._sdp, 10) == NotFound def test_i_can_get_an_retrieve_value_from_cache(self): cache = Cache() cache.put("key", "value") assert cache.get("key") == "value" assert len(cache) == 1 cache.put("key", "another value") # another value in the cache replace the previous one assert cache.get("key") == "another value" assert len(cache) == 1 cache.put("key2", "value2") # another key assert cache.get("key2") == "value2" assert len(cache) == 2 assert cache.copy() == {"key": "another value", "key2": "value2"} def test_i_do_not_evict_when_put(self): """ It's because we evict on get() :return: :rtype: """ maxsize = 5 cache = Cache(max_size=5) for key in range(maxsize + 2): cache.put(str(key), key) assert len(cache) == maxsize + 2 assert cache.copy() == { "0": 0, "1": 1, "2": 2, "3": 3, "4": 4, "5": 5, "6": 6, } def test_i_can_evict_when_get(self): maxsize = 5 cache = Cache(max_size=5, default=lambda k: int(k)) for key in range(maxsize + 2): cache.get(str(key)) assert len(cache) == maxsize assert cache.copy() == { "2": 2, "3": 3, "4": 4, "5": 5, "6": 6, } def test_i_do_not_evict_when_items_are_not_committed(self): maxsize = 5 cache = Cache(max_size=5, default=lambda k: k) for key in range(maxsize + 2): cache.put(str(key), key) assert len(cache) == maxsize + 2 cache.get("-1") assert len(cache) == maxsize + 2 assert cache.copy() == { "0": 0, "1": 1, "2": 2, "3": 3, "4": 4, "5": 5, "6": 6, } def test_i_can_get_a_value_from_alt_sdp(self): cache = Cache(sdp=FakeSdp(get_value=lambda cache_name, key: NotFound)).auto_configure("cache_name") alt_sdp = FakeSdp(get_alt_value=lambda cache_name, key: "value found !") assert cache.get("key", alt_sdp=alt_sdp) == "value found !" # The value is now in cache assert cache.copy() == {'key': 'value found !'} def test_i_cannot_get_a_value_from_alt_sdp_when_cache_is_cleared(self): cache = Cache(sdp=FakeSdp(get_value=lambda cache_name, key: NotFound)).auto_configure("cache_name") cache.clear() alt_sdp = FakeSdp(get_alt_value=lambda cache_name, key: "value found !") assert cache.get("key", alt_sdp=alt_sdp) is NotFound assert cache.copy() == {} def test_i_can_get_default_value_from_simple_cache(self): cache = Cache() assert cache.get("key") is NotFound cache = Cache(default=10) assert cache.get("key") == 10 assert "key" not in cache # default value are not put in cache cache = Cache(default=lambda key: key + "_not_found") assert cache.get("key") == "key_not_found" assert "key" in cache # default callable are put in cache cache = Cache(default=lambda sdp, key: sdp.get("cache_name", key), sdp=FakeSdp(get_value=lambda entry, key: key + "_not_found")) assert cache.get("key") == "key_not_found" assert "key" in cache # default callable are put in cache def test_i_do_not_ask_the_remote_repository_twice(self): nb_request = [] cache = Cache(default=lambda key: nb_request.append("requested")) assert cache.get("key") is None assert cache.get("key") is None assert len(nb_request) == 1 def test_i_can_update_from_simple_cache(self): cache = Cache() cache.put("key", "value") cache.update("key", "value", "key", "new_value") assert len(cache._cache) == 1 assert len(cache) == 1 assert cache.get("key") == "new_value" assert cache.to_add == {"key"} assert cache.to_remove == set() cache.reset_events() cache.update("key", "new_value", "another_key", "another_value") assert len(cache._cache) == 1 assert len(cache) == 1 assert cache.get("key") is NotFound assert cache.get("another_key") == "another_value" assert cache.to_add == {"another_key"} assert cache.to_remove == {"key"} with pytest.raises(KeyError): cache.update("wrong key", "value", "key", "value") def test_i_can_update_when_alt_sdp_same_keys(self): cache = Cache(default=lambda sdp, key: sdp.get("cache_name", key), extend_exists=lambda sdp, key: sdp.exists("cache_name", key), sdp=FakeSdp(get_value=lambda cache_name, key: NotFound)) cache.put("key", "value") cache.update("key", "value", "key", "new_value", FakeSdp(extend_exists=lambda cache_name, key: True)) assert cache.get("key") == "new_value" def test_i_can_update_when_alt_sdp_different_keys(self): cache = Cache(default=lambda sdp, key: sdp.get("cache_name", key), extend_exists=lambda sdp, key: sdp.exists("cache_name", key), sdp=FakeSdp(get_value=lambda cache_name, key: NotFound)) cache.put("key", "value") cache.update("key", "value", "key2", "value2", FakeSdp(extend_exists=lambda cache_name, key: True)) assert cache.get("key2") == "value2" assert cache.get("key") == Removed assert cache.to_add == {"key", "key2"} assert cache.to_remove == set() @pytest.mark.parametrize("cache", [ Cache(), ListCache(), ListIfNeededCache(), SetCache(), IncCache() ]) def test_i_can_manage_cache_events(self, cache: BaseCache): cache.put("key", "value") assert cache.to_add == {"key"} assert cache.to_remove == set() cache.update("key", "value", "key", "another value") assert cache.to_add == {"key"} assert cache.to_remove == set() cache.update("key", "another value", "key2", "value2") assert cache.to_add == {"key2"} assert cache.to_remove == {"key"} cache.update("key2", "value2", "key", "value") assert cache.to_add == {"key"} assert cache.to_remove == {"key2"} @pytest.mark.parametrize("cache", [ ListCache(), SetCache(), ListIfNeededCache() ]) def test_i_can_manage_list_and_set_cache_events(self, cache): cache.put("key", "value") cache.put("key", "value2") assert cache.to_add == {"key"} assert cache.to_remove == set() cache.update("key", "value", "key", "another value") assert cache.to_add == {"key"} assert cache.to_remove == set() cache.update("key", "value2", "key2", "value2") assert cache.to_add == {"key", "key2"} assert cache.to_remove == set() cache.update("key", "another value", "key3", "another value") assert cache.to_add == {"key2", "key3"} assert cache.to_remove == {"key"} @pytest.mark.parametrize("cache", [ Cache(), ListCache(), SetCache(), ListIfNeededCache(), IncCache() ]) def test_exists(self, cache): assert not cache.exists("key") cache.put("key", "value") assert cache.exists("key") def test_exists_extend(self): cache = Cache(extend_exists=lambda k: True if k == "special_key" else False) assert not cache.exists("key") assert cache.exists("special_key") def test_i_can_extend_exists_when_internal_sdp(self): cache = Cache(extend_exists=lambda sdp, k: True if k == "special_key" else False, sdp=FakeSdp) assert not cache.exists("key") assert cache.exists("special_key") @pytest.mark.parametrize("cache, default, new_value, expected", [ (ListCache(), lambda k: NotFound, "value", ["value"]), (ListCache(), lambda k: ["value"], "value", ["value", "value"]), (ListIfNeededCache(), lambda k: NotFound, "value", "value"), (ListIfNeededCache(), lambda k: "value", "value1", ["value", "value1"]), (ListIfNeededCache(), lambda k: ["value1", "value2"], "value1", ["value1", "value2", "value1"]), (SetCache(), lambda k: NotFound, "value", {"value"}), (SetCache(), lambda k: {"value"}, "value", {"value"}), (SetCache(), lambda k: {"value1"}, "value2", {"value1", "value2"}), ]) def test_default_is_called_before_put_to_keep_in_sync(self, cache, default, new_value, expected): cache.configure(default=default) cache.put("key", new_value) assert cache.get("key") == expected def test_default_is_called_before_updating_simple_cache(self): cache = Cache(default=lambda k: NotFound) with pytest.raises(KeyError): cache.update("old_key", "old_value", "new_key", "new_value") cache = Cache(default=lambda k: "old_value") cache.update("old_key", "old_value", "new_key", "new_value") assert cache.get("new_key") == "new_value" def test_i_can_delete_an_entry_from_cache(self): cache = Cache() cache.put("key", "value") assert cache.get("key") == "value" cache.delete("key") assert cache.get("key") is NotFound assert cache.to_remove == {"key"} def test_i_can_delete_when_entry_is_only_in_db(self): cache = Cache(default=lambda k: "value" if k == 'key' else NotFound) cache.delete("another_key") assert cache.copy() == {} assert cache.to_add == set() assert cache.to_remove == set() cache.delete("key") assert cache.copy() == {} assert cache.to_add == set() assert cache.to_remove == {"key"} def test_i_can_delete_an_entry_from_cache_when_alt_sdp_and_value_in_cache(self): # There is a value in alt_cache_manager, # No remaining value in current cache after deletion # The key must be flagged as Removed cache = Cache(extend_exists=lambda sdp, k: sdp.exists("cache_name", k)) cache.put("key", "value") cache.delete("key", value=None, alt_sdp=FakeSdp(extend_exists=lambda cache_name, key: True)) assert cache.copy() == {"key": Removed} assert cache.to_add == {"key"} assert cache.to_remove == set() def test_i_can_delete_an_entry_from_cache_when_alt_sdp_when_in_remote_repository(self): # There is a value in alt_cache_manager, # No remaining value in current cache after deletion # The key must be flagged as Removed cache = Cache(default=lambda k: "value", extend_exists=lambda sdp, k: sdp.exists("cache_name", k)) cache.delete("key", value=None, alt_sdp=FakeSdp(extend_exists=lambda cache_name, key: True)) assert cache.copy() == {"key": Removed} assert cache.to_add == {"key"} assert cache.to_remove == set() def test_i_can_delete_an_entry_from_cache_when_alt_sdp_and_no_value_in_cache_or_remote_repository(self): # alt_cache_manager is used when no value found cache = Cache(default=lambda sdp, k: sdp.get("cache_name", k), extend_exists=lambda sdp, k: sdp.exists("cache_name", k), sdp=FakeSdp(get_value=lambda entry, k: NotFound)) cache.delete("key", value=None, alt_sdp=FakeSdp(extend_exists=lambda cache_name, key: True)) assert cache.copy() == {"key": Removed} assert cache.to_add == {"key"} assert cache.to_remove == set() def test_no_error_when_deleting_a_key_that_does_not_exists_when_alt_sdp(self): # alt_cache_manager is used when no value found cache = Cache(default=lambda sdp, k: sdp.get("cache_name", k), extend_exists=lambda sdp, k: sdp.exists("cache_name", k), sdp=FakeSdp(get_value=lambda entry, k: NotFound)) cache.delete("key", value=None, alt_sdp=FakeSdp(extend_exists=lambda cache_name, key: False)) assert cache.copy() == {} assert cache.to_add == set() assert cache.to_remove == set() def test_initialized_key_is_removed_when_the_entry_is_found(self): caches = [Cache(), ListCache(), ListIfNeededCache(), SetCache()] for cache in caches: cache.put("key", "value") cache.get("key") assert len(cache._initialized_keys) == 0 cache = IncCache() cache.put("key", 10) cache.get("key") assert len(cache._initialized_keys) == 0 def test_initialized_keys_are_reset_when_max_length_is_reached(self): cache = Cache() for i in range(MAX_INITIALIZED_KEY): cache.get(str(i)) assert len(cache._initialized_keys) == MAX_INITIALIZED_KEY cache.get(str(MAX_INITIALIZED_KEY + 1)) assert len(cache._initialized_keys) == 1 def test_i_can_populate(self): items = [("1", "1"), ("2", "2"), ("3", "3")] cache = Cache() cache.populate(lambda: items, lambda item: item[0]) assert len(cache) == 3 assert cache.get("1") == ("1", "1") assert cache.get("2") == ("2", "2") assert cache.get("3") == ("3", "3") assert cache.to_add == {"1", "2", "3"} assert cache.to_remove == set() def test_i_can_populate_using_internal_sdp(self): items = [("1", "1"), ("2", "2"), ("3", "3")] cache = Cache(sdp=FakeSdp(populate=items)) cache.populate(lambda sdp: sdp.populate(), lambda item: item[0]) assert len(cache) == 3 assert cache.get("1") == ("1", "1") assert cache.get("2") == ("2", "2") assert cache.get("3") == ("3", "3") assert cache.to_add == {"1", "2", "3"} assert cache.to_remove == set() def test_i_can_reset_the_event_after_populate(self): items = [("1", "1"), ("2", "2"), ("3", "3")] cache = Cache() cache.to_add = {"some_value"} cache.to_remove = {"some_other_value"} cache.populate(lambda: items, lambda item: item[0], reset_events=True) assert len(cache) == 3 assert cache.copy() == {"1": ("1", "1"), "2": ("2", "2"), "3": ("3", "3")} assert cache.to_add == {"some_value"} assert cache.to_remove == {"some_other_value"} def test_i_can_get_all(self): items = [("1", "1"), ("2", "2"), ("3", "3")] cache = Cache() cache.populate(lambda: items, lambda item: item[0]) res = cache.get_all() assert len(res) == 3 assert list(res) == [('1', '1'), ('2', '2'), ('3', '3')] def test_i_can_clone_cache(self): cache = Cache(max_size=256, default=lambda sdp, key: sdp.get("cache_name", key), extend_exists=False, alt_sdp_get=lambda sdp, key: sdp.alt_get("cache_name", key), sdp=FakeSdp(get_value=lambda entry, key: key + "_not_found")) cache.put("key1", "value1") cache.put("key2", "value2") clone = cache.clone() assert type(cache) == type(clone) assert clone._max_size == cache._max_size assert clone._default == cache._default assert clone._extend_exists == cache._extend_exists assert clone._alt_sdp_get == cache._alt_sdp_get assert clone._sdp == cache._sdp assert clone._cache == {} # value are not copied assert clone._initialized_keys == set() assert clone._current_size == 0 assert clone.to_add == set() assert clone.to_remove == set() clone.configure(sdp=FakeSdp(lambda entry, key: key + " found !")) assert cache.get("key3") == "key3_not_found" assert clone.get("key3") == "key3 found !" @pytest.mark.parametrize("cache", [ Cache(), DictionaryCache(), IncCache(), ListCache(), ListIfNeededCache() ]) def test_i_can_clone_all_caches(self, cache): clone = cache.clone() assert type(clone) == type(cache) def test_sanity_check_on_list_if_needed_cache(self): cache = ListIfNeededCache() clone = cache.clone() clone.put("key", "value1") clone.put("key", "value2") assert clone.get("key") == ["value1", "value2"] def test_i_can_clear_when_alt_sdp(self): cache = Cache().auto_configure("cache_name") cache.put("key1", "value1") cache.put("key2", "value2") cache.clear() assert cache.copy() == {} assert cache._is_cleared def test_i_can_iter_on_the_content(self): cache = Cache() cache.put("key1", "value1") cache.put("key2", "value2") cache.put("key3", "value3") res = [] for k in cache: assert k in cache res.append(k) assert res == ["key1", "key2", "key3"]