from dataclasses import dataclass import pytest from base import BaseTest from caching.Cache import Cache from caching.CacheManager import CacheManager, ConceptNotFound from caching.ListIfNeededCache import ListIfNeededCache from common.global_symbols import NotFound from helpers import get_metadata from tests.caching import FakeSdp @dataclass class Obj: key: str value: str class TestCacheManager(BaseTest): def test_i_can_push_into_sdp(self, context, sdp): cache_manager = CacheManager(sdp) cache_manager.register_cache("test", Cache(), persist=True) cache_manager.put("test", "key", "value") cache_manager.commit(context) assert sdp.exists("test", "key") def test_sdp_given_to_the_cache_manager_can_be_used_by_the_cache(self, context, sdp): """ When the sdp is given to the cache_manager, the 'Cache' object can use the lambda with two parameters :param context: :type context: :return: :rtype: """ cache_manager = CacheManager(sdp) cache = Cache(default=lambda _sdp, k: _sdp.get("test", k)) cache_manager.register_cache("test", cache) with cache_manager.sdp.get_transaction(context.event) as transaction: transaction.add("test", "key", "value") assert cache_manager.get("test", "key") == "value" def test_cache_can_use_auto_configure_when_sdp_is_given_to_the_cache_manager(self, context, sdp): """ When sdp is given to the cache manager, Cache can be configured using auto_configure as the sdp will be provided by the cache_manager :param context: :type context: :return: :rtype: """ with sdp.get_transaction(context.event) as transaction: transaction.add("test", "key", "value") cache_manager = CacheManager(sdp) cache = Cache().auto_configure("test") cache_manager.register_cache("test", cache) assert cache_manager.get("test", "key") == "value" def test_i_can_get_value_when_the_sdp_is_given_to_the_cache(self, context, sdp): """ If the sdp is not given to the cache manager, the cache will need to explicitly provide the sdp during its configuration :param context: :type context: :return: :rtype: """ with sdp.get_transaction(context.event) as transaction: transaction.add("test", "key", "value") cache_manager = CacheManager() cache = Cache(default=lambda k: sdp.get("test", k)) cache_manager.register_cache("test", cache) assert cache_manager.get("test", "key") == "value" def test_i_can_get_value_from_alt_sdp(self, sdp): """ When nothing is found in cache and in sdp, we use alternate sdp :param sdp: :type sdp: :return: :rtype: """ cache_manager = CacheManager(sdp) cache_manager.register_cache("test", Cache().auto_configure("test")) alt_sdp = FakeSdp(get_alt_value=lambda cache_name, key: "value found !") assert cache_manager.get("test", "key", alt_sdp=alt_sdp) == "value found !" def test_i_can_commit_simple_cache(self, context, sdp): cache_manager = CacheManager(sdp) cache_manager.register_cache("test", Cache().auto_configure("test")) cache = cache_manager.caches["test"].cache cache_manager.put("test", "key", "value") cache_manager.commit(context) cache.clear() assert cache_manager.sdp.get("test", "key") == "value" assert cache.get("key") == "value" cache.update("key", "value", "key", "another_value") cache_manager.commit(context) assert cache_manager.sdp.get("test", "key") == "another_value" cache.update("key", "another_value", "key2", "another_value") cache_manager.commit(context) assert cache_manager.sdp.get("test", "key") is NotFound assert cache_manager.sdp.get("test", "key2") == "another_value" # sanity check # sdp 'test' has value, but sdp '__default__' does not assert cache_manager.sdp.name == "test" assert cache_manager.sdp.state.data == {'test': {'key2': 'another_value'}} def test_i_can_use_concept_cache(self): cache_manager = CacheManager() cache_manager.register_concept_cache("id", Cache(), lambda c: c.id, True) cache_manager.register_concept_cache("name", ListIfNeededCache(), lambda c: c.name, True) # caches are correctly created assert cache_manager.concept_caches == ["id", "name"] assert "id" in cache_manager.caches assert "name" in cache_manager.caches # caches are correctly updated on insertion meta1 = get_metadata(id="1", name="foo") meta2 = get_metadata(id="2", name="bar") meta3 = get_metadata(id="3", name="foo") for metadata in meta1, meta2, meta3: cache_manager.add_concept(metadata) assert cache_manager.get_inner_cache("id").copy() == {"1": meta1, "2": meta2, "3": meta3} assert cache_manager.get_inner_cache("name").copy() == {"foo": [meta1, meta3], "bar": meta2} # caches are correctly updated on modification meta3prime = get_metadata(id="3", name="bar") cache_manager.update_concept(meta3, meta3prime) assert cache_manager.get_inner_cache("id").copy() == {"1": meta1, "2": meta2, "3": meta3prime} assert cache_manager.get_inner_cache("name").copy() == {"foo": meta1, "bar": [meta2, meta3prime]} # caches are correctly updated on removal cache_manager.remove_concept(meta3prime) assert cache_manager.get_inner_cache("id").copy() == {"1": meta1, "2": meta2} assert cache_manager.get_inner_cache("name").copy() == {"foo": meta1, "bar": meta2} def test_i_cannot_remove_a_concept_that_does_not_exists(self): cache_manager = CacheManager() cache_manager.register_concept_cache("id", Cache(), lambda c: c.id, True) cache_manager.register_concept_cache("key", ListIfNeededCache(), lambda c: c.key, True) meta1 = get_metadata(id="1", name="foo") with pytest.raises(ConceptNotFound) as ex: cache_manager.remove_concept(meta1) assert ex.value.concept == meta1 def test_nothing_is_sent_to_sdp_if_persist_is_false(self, context, sdp): cache_manager = CacheManager(sdp) cache_manager.register_cache("test", Cache(), persist=False) cache_manager.put("test", "key", "value") cache_manager.commit(context) value = sdp.get("test", "key") assert value is NotFound def test_i_can_delete_from_cache_manager(self, context, sdp): cache_manager = CacheManager(sdp) cache_manager.register_cache("test", Cache(), persist=True) cache_manager.put("test", "key", "value") cache_manager.commit(context) # sanity check assert cache_manager.get("test", "key") == "value" assert sdp.get("test", "key") == "value" # I remove but I don't commit cache_manager.delete("test", "key") assert cache_manager.get("test", "key") is NotFound assert sdp.get("test", "key") == "value" # commit cache_manager.commit(context) assert cache_manager.get("test", "key") is NotFound assert sdp.get("test", "key") is NotFound def test_i_can_get_the_inner_cache(self): cache_manager = CacheManager() cache = Cache() cache_manager.register_cache("test", cache) inner_cache = cache_manager.get_inner_cache("test") assert id(inner_cache) == id(cache) def test_i_can_get_a_copy_of_a_cache(self): cache_manager = CacheManager() cache = Cache() cache.put("key1", "value1") cache.put("key2", "value2") cache_manager.register_cache("test", cache) copy = cache_manager.copy("test") assert isinstance(copy, dict) assert copy == {"key1": "value1", "key2": "value2"} def test_i_can_populate(self): cache_manager = CacheManager() cache = Cache() cache_manager.register_cache("test", cache) obj1, obj2 = Obj("key1", "value1"), Obj("key2", "value2") cache_manager.populate("test", lambda: [obj1, obj2], lambda o: o.key) assert cache_manager.copy("test") == {"key1": obj1, "key2": obj2} def test_has(self, context, sdp): cache_manager = CacheManager(sdp) cache_manager.register_cache("test", Cache().auto_configure("test")) with cache_manager.sdp.get_transaction(context.event) as transaction: transaction.add("test", "key_from_sdp", "value") cache_manager.put("test", "key_in_cache", "value") assert cache_manager.has("test", "key_in_cache") assert not cache_manager.has("test", "key_from_sdp") def test_exist(self, context, sdp): cache_manager = CacheManager(sdp) cache_manager.register_cache("test", Cache().auto_configure("test")) with cache_manager.sdp.get_transaction(context.event) as transaction: transaction.add("test", "key_from_sdp", "value") cache_manager.put("test", "key_in_cache", "value") assert cache_manager.exists("test", "key_in_cache") assert cache_manager.exists("test", "key_from_sdp") def test_i_can_clear_a_single_cache(self, context, sdp): cache_manager = CacheManager(sdp) cache_manager.register_cache("test", Cache().auto_configure("test")) cache_manager.put("test", "key1", "value1") cache_manager.put("test", "key2", "value2") cache_manager.commit(context) cache_manager.clear("test") assert cache_manager.copy("test") == {} assert sdp.exists("test", "key1") assert sdp.exists("test", "key2") cache_manager.commit(context) assert cache_manager.copy("test") == {} assert not sdp.exists("test", "key1") assert not sdp.exists("test", "key2") def test_i_can_clear_all_caches(self, sdp): cache_manager = CacheManager(sdp) cache_manager.register_cache("test1", Cache()) cache_manager.register_cache("test2", Cache()) cache_manager.put("test1", "key", "value") cache_manager.put("test2", "key", "value") cache_manager.clear() assert cache_manager.copy("test1") == {} assert cache_manager.copy("test2") == {} def test_i_cannot_add_null_keys_into_concept_cache(self): cache_manager = CacheManager() cache_manager.register_concept_cache("id", Cache(), lambda c: c.id, True) with pytest.raises(KeyError): meta1 = get_metadata(name="foo") cache_manager.add_concept(meta1)