from dataclasses import dataclass import pytest from cache.Cache import Cache from cache.DictionaryCache import DictionaryCache from cache.IncCache import IncCache from cache.ListCache import ListCache from cache.ListIfNeededCache import ListIfNeededCache from core.concept import Concept from core.global_symbols import NotFound, Removed, EVENT_CONCEPT_ID_DELETED, \ EVENT_RULE_ID_DELETED from core.rule import Rule, ACTION_TYPE_EXEC from core.sheerka.SheerkaOntologyManager import SheerkaOntologyManager, OntologyManagerFrozen, OntologyManagerNotFrozen, \ OntologyManagerCannotPopLatest, OntologyAlreadyExists from tests.TestUsingFileBasedSheerka import TestUsingFileBasedSheerka from tests.TestUsingMemoryBasedSheerka import TestUsingMemoryBasedSheerka @dataclass class DummyObj: key: str value: object class TestSheerkaOntology(TestUsingMemoryBasedSheerka): def test_i_can_create_ontology_manager(self): sheerka = self.get_sheerka() manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) assert len(manager.ontologies) == 1 assert id(manager.current_cache_manager()) == id(manager.ontologies[0].cache_manager) assert id(manager.current_sdp()) == id(manager.ontologies[0].cache_manager.sdp) def test_i_can_get_value_from_the_current_cache_manager(self): sheerka, context = self.init_concepts(cache_only=False) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) cache = Cache() manager.register_cache("test", cache) assert cache._sdp == manager.current_sdp() manager.put("test", "key", "value") assert manager.get("test", "key") == "value" assert manager.current_sdp().get("test", "key") == NotFound manager.commit(context) assert manager.get("test", "key") == "value" assert manager.current_sdp().get("test", "key") == "value" def test_i_cannot_register_cache_once_ontology_is_frozen(self): sheerka = self.get_sheerka() manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.freeze() with pytest.raises(OntologyManagerFrozen): manager.register_cache("test", Cache()) with pytest.raises(OntologyManagerFrozen): manager.register_concept_cache("test", Cache(), lambda obj: obj.key, True) def test_i_cannot_push_ontology_if_not_frozen(self): sheerka = self.get_sheerka() ontology = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) with pytest.raises(OntologyManagerNotFrozen): ontology.push_ontology("new_ontology") def test_i_can_push_ontology_from_simple_caches(self): sheerka = self.get_sheerka() manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("Cache", Cache(), persist=True, use_ref=True) manager.register_cache("DictionaryCache", DictionaryCache(), persist=True, use_ref=False) manager.register_cache("ListIfNeededCache", ListIfNeededCache(), persist=False, use_ref=False) manager.register_cache("ListCache", ListCache(), persist=False, use_ref=True) manager.register_cache("IncCache", IncCache(), False) manager.freeze() manager.push_ontology("new_ontology") assert len(manager.ontologies) == 2 cache_manager_0 = manager.ontologies[0].cache_manager assert len(cache_manager_0.caches) == 5 assert cache_manager_0.concept_caches == [] assert isinstance(cache_manager_0.get_cache("Cache"), Cache) assert cache_manager_0.caches["Cache"].persist == True assert cache_manager_0.caches["Cache"].use_ref == True assert cache_manager_0.get_cache("Cache")._sdp.name == "new_ontology" assert isinstance(cache_manager_0.get_cache("DictionaryCache"), DictionaryCache) assert cache_manager_0.caches["DictionaryCache"].persist == True assert cache_manager_0.caches["DictionaryCache"].use_ref == False assert cache_manager_0.get_cache("DictionaryCache")._sdp.name == "new_ontology" assert isinstance(cache_manager_0.get_cache("ListIfNeededCache"), ListIfNeededCache) assert cache_manager_0.caches["ListIfNeededCache"].persist == False assert cache_manager_0.caches["ListIfNeededCache"].use_ref == False assert cache_manager_0.get_cache("ListIfNeededCache")._sdp.name == "new_ontology" assert isinstance(cache_manager_0.get_cache("ListCache"), ListCache) assert cache_manager_0.caches["ListCache"].persist == False assert cache_manager_0.caches["ListCache"].use_ref == True assert cache_manager_0.get_cache("ListCache")._sdp.name == "new_ontology" assert isinstance(cache_manager_0.get_cache("IncCache"), IncCache) assert cache_manager_0.sdp.name == "new_ontology" assert cache_manager_0.get_cache("IncCache")._sdp.name == "new_ontology" # old ontology is still there cache_manager_1 = manager.ontologies[1].cache_manager assert len(cache_manager_1.caches) == 5 assert isinstance(cache_manager_1.get_cache("Cache"), Cache) assert isinstance(cache_manager_1.get_cache("DictionaryCache"), DictionaryCache) assert isinstance(cache_manager_1.get_cache("ListIfNeededCache"), ListIfNeededCache) assert isinstance(cache_manager_1.get_cache("ListCache"), ListCache) assert isinstance(cache_manager_1.get_cache("IncCache"), IncCache) assert cache_manager_1.sdp.name == "__default__" def test_i_can_push_ontology_from_concept_caches(self): sheerka = self.get_sheerka() manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_concept_cache("ByKey", Cache(), get_key=lambda obj: obj.key, use_ref=True) manager.register_concept_cache("ById", ListIfNeededCache(), get_key=lambda obj: obj.id, use_ref=False) manager.freeze() manager.push_ontology("new_ontology") assert len(manager.ontologies) == 2 cache_manager_0 = manager.ontologies[0].cache_manager cache_manager_1 = manager.ontologies[1].cache_manager assert len(cache_manager_0.caches) == 2 assert isinstance(cache_manager_0.get_cache("ByKey"), Cache) assert cache_manager_0.caches["ByKey"].persist == cache_manager_1.caches["ByKey"].persist assert cache_manager_0.caches["ByKey"].get_key == cache_manager_1.caches["ByKey"].get_key assert cache_manager_0.caches["ByKey"].use_ref == cache_manager_1.caches["ByKey"].use_ref assert isinstance(cache_manager_0.get_cache("ById"), ListIfNeededCache) assert cache_manager_0.caches["ById"].persist == cache_manager_1.caches["ById"].persist assert cache_manager_0.caches["ById"].get_key == cache_manager_1.caches["ById"].get_key assert cache_manager_0.caches["ById"].use_ref == cache_manager_1.caches["ById"].use_ref assert cache_manager_0.concept_caches == cache_manager_1.concept_caches assert cache_manager_0.sdp.name == "new_ontology" assert cache_manager_1.sdp.name == "__default__" def test_i_can_get_database_value(self): sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache(default=lambda sdp, key: sdp.get("cache_name", key))) with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", "value") assert not manager.current_cache_manager().has("cache_name", "key") assert manager.get("cache_name", "key") == "value" assert manager.current_cache_manager().has("cache_name", "key") def test_i_cannot_pop_ontology_when_not_frozen(self): sheerka, context = self.init_test().unpack() manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) with pytest.raises(OntologyManagerNotFrozen): manager.pop_ontology(context) def test_i_cannot_pop_the_latest_cache_manager(self): sheerka, context = self.init_test().unpack() manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.freeze() with pytest.raises(OntologyManagerCannotPopLatest): manager.pop_ontology(context) def test_i_can_pop_ontology(self): sheerka, context = self.init_test().unpack() manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.freeze() manager.push_ontology("ontology1") manager.push_ontology("ontology2") manager.push_ontology("ontology3") manager.pop_ontology(context) assert len(manager.ontologies) == 3 manager.pop_ontology(context) manager.pop_ontology(context) with pytest.raises(OntologyManagerCannotPopLatest): manager.pop_ontology(context) def test_i_can_add_ontology(self): sheerka, context = self.init_test().unpack() manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache().auto_configure("cache_name")) manager.freeze() # init the ontology that will be put back manager.push_ontology("new ontology") manager.put("cache_name", "key1", "value1") manager.commit(context) with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key2", "value2") # call key3 to check. This time there is no value, but there will be later assert manager.get("cache_name", "key3") is NotFound new_ontology = manager.get_ontology() manager.pop_ontology(context) # add another ontology, with its own values manager.push_ontology("another ontology") manager.put("cache_name", "key1", "value1_from_another") manager.put("cache_name", "key2", "value2_from_another") manager.put("cache_name", "key3", "value3_from_another") manager.commit(context) # put back the ontology manager.add_ontology(new_ontology) assert manager.get("cache_name", "key1") == "value1" assert manager.get("cache_name", "key2") == "value2" assert manager.get("cache_name", "key3") == "value3_from_another" def test_i_can_get_ontology(self): sheerka, context = self.init_test().unpack() manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.freeze() manager.push_ontology("name1") manager.push_ontology("name2") manager.push_ontology("name3") assert manager.get_ontology("name2").name == "name2" assert manager.get_ontology().name == "name3" with pytest.raises(KeyError): assert manager.get_ontology("name4") def test_i_can_access_values_after_push_and_pop_cache_only_true(self): sheerka, context = self.init_test(cache_only=True).unpack() manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache()) manager.freeze() manager.put("cache_name", "key", "value1") assert manager.get("cache_name", "key") == "value1" manager.push_ontology("new ontology") manager.put("cache_name", "key", "value2") assert manager.get("cache_name", "key") == "value2" manager.pop_ontology(context) assert manager.get("cache_name", "key") == "value1" def test_i_can_access_values_after_push_and_pop_cache_only_false(self): sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache(default=lambda sdp, key: sdp.get("cache_name", key))) manager.freeze() # put value in DB with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", "value1") assert not manager.current_cache_manager().has("cache_name", "key") # value not in cache assert manager.current_sdp().exists("cache_name", "key") # but value is in DB # add an ontology layer manager.push_ontology("new ontology") with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", "value2") # At this point, the value is in DB, but not in cache assert not manager.current_cache_manager().has("cache_name", "key") assert manager.get("cache_name", "key") == "value2" assert manager.current_cache_manager().has("cache_name", "key") # sanity check # Let's check sdp values assert manager.ontologies[0].cache_manager.sdp.state.data == {'cache_name': {'key': 'value2'}} assert manager.ontologies[1].cache_manager.sdp.state.data == {'cache_name': {'key': 'value1'}} # remove a layer manager.pop_ontology(context) assert not manager.current_cache_manager().has("cache_name", "key") # value is no longer in cache assert manager.get("cache_name", "key") == "value1" # sanity check # Let's check sdp values assert manager.ontologies[0].cache_manager.sdp.state.data == {'cache_name': {'key': 'value1'}} def test_i_can_manage_multiple_ontology_layers(self): sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache(default=lambda sdp, key: sdp.get("cache_name", key))) manager.freeze() # default layer with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", "value1") # add an ontology layer manager.push_ontology("new ontology") with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", "value2") # add an ontology layer manager.push_ontology("another ontology") with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", "value3") # add an ontology layer manager.push_ontology("fourth ontology") with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", "value4") assert manager.get("cache_name", "key") == "value4" manager.pop_ontology(context) assert manager.get("cache_name", "key") == "value3" manager.pop_ontology(context) assert manager.get("cache_name", "key") == "value2" manager.pop_ontology(context) assert manager.get("cache_name", "key") == "value1" def test_i_have_access_to_sub_layers_values_cache_only_false(self): sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache().auto_configure("cache_name")) manager.freeze() # default layer with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", "value") # add ontology layers manager.push_ontology("new ontology") manager.push_ontology("another ontology") # I can get the low level value assert manager.get("cache_name", "key") == "value" # check that the value is copied on the top level cache assert manager.current_cache_manager().has("cache_name", "key") assert not manager.ontologies[1].cache_manager.has("cache_name", "key") # not the top level assert manager.ontologies[2].cache_manager.has("cache_name", "key") # the data comes from it def test_i_can_get_value_from_all_layers(self): sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache().auto_configure("cache_name")) manager.freeze() manager.put("cache_name", "key", "value") # add ontology layers manager.push_ontology("new ontology") manager.push_ontology("another ontology") assert manager.get("cache_name", "key") == "value" manager.pop_ontology(context) assert manager.get("cache_name", "key") == "value" manager.pop_ontology(context) assert manager.get("cache_name", "key") == "value" def test_i_can_only_get_top_layer_values_when_dictionary_cache(self): sheerka = self.get_sheerka() context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", DictionaryCache().auto_configure("cache_name")) manager.freeze() manager.put("cache_name", False, {"key": "value"}) # add some values in default layer # add some other values in another layer manager.push_ontology("new ontology") manager.put("cache_name", False, {"key1": "value1"}) assert manager.get("cache_name", "key") is NotFound # other layer are not visible assert manager.get("cache_name", "key1") == "value1" # I an only see the current layer # I still can use get all assert manager.get_all("cache_name") == {"key": "value", "key1": "value1"} # I can get back my values after pop manager.pop_ontology(context) assert manager.copy("cache_name") == {"key": "value"} def test_dictionary_caches_values_are_copied_when_a_new_ontology_is_pushed(self): sheerka = self.get_sheerka() manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", DictionaryCache().auto_configure("cache_name")) manager.freeze() manager.put("cache_name", False, {"key": "value"}) # add some values in default layer manager.push_ontology("new ontology") assert manager.copy("cache_name") == {"key": "value"} assert manager.current_cache_manager().get_cache("cache_name").to_add == set() assert manager.current_cache_manager().get_cache("cache_name").to_remove == set() def test_initialized_key_are_correctly_managed_when_multiple_layers(self): sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("c_name", Cache().auto_configure("c_name")) manager.freeze() manager.put("c_name", "key", "value") with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("c_name", "key2", "value2") # not in cache # add ontology layers manager.push_ontology("new ontology") manager.push_ontology("another ontology") manager.push_ontology("last ontology") manager.get("c_name", "key") # == "value" but we don't care assert manager.ontologies[0].cache_manager.caches["c_name"].cache._initialized_keys == {"key"} assert manager.ontologies[1].cache_manager.caches["c_name"].cache._initialized_keys == set() assert manager.ontologies[2].cache_manager.caches["c_name"].cache._initialized_keys == set() assert manager.ontologies[3].cache_manager.caches["c_name"].cache._initialized_keys == set() manager.get("c_name", "key2") # == "value2" but we don't care assert manager.ontologies[0].cache_manager.caches["c_name"].cache._initialized_keys == {"key", "key2"} assert manager.ontologies[1].cache_manager.caches["c_name"].cache._initialized_keys == set() assert manager.ontologies[2].cache_manager.caches["c_name"].cache._initialized_keys == set() assert manager.ontologies[3].cache_manager.caches["c_name"].cache._initialized_keys == {"key2"} manager.get("c_name", "no_key") # is NotFound but we don't care assert manager.ontologies[0].cache_manager.caches["c_name"].cache._initialized_keys == {"key", "key2", "no_key"} assert manager.ontologies[1].cache_manager.caches["c_name"].cache._initialized_keys == set() assert manager.ontologies[2].cache_manager.caches["c_name"].cache._initialized_keys == set() assert manager.ontologies[3].cache_manager.caches["c_name"].cache._initialized_keys == {"key2", "no_key"} def test_i_cannot_get_a_value_that_does_not_exists(self): sheerka = self.get_sheerka(cache_only=False) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache().auto_configure("cache_name")) manager.freeze() # add ontology layers manager.push_ontology("new ontology") manager.push_ontology("another ontology") assert manager.get("cache_name", "key") is NotFound def test_i_cannot_get_a_value_that_is_removed(self): sheerka = self.get_sheerka() manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache()) manager.freeze() manager.put("cache_name", "key", "value") # add ontology layers manager.push_ontology("new ontology") manager.put("cache_name", "key", Removed) assert manager.get("cache_name", "key") is NotFound def test_i_cannot_get_value_that_is_removed_in_sub_level(self): sheerka = self.get_sheerka(cache_only=False) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache().auto_configure("cache_name")) manager.freeze() manager.put("cache_name", "key", "value") # value exists # add ontology layer manager.push_ontology("new ontology") manager.put("cache_name", "key", Removed) # value is removed # add another layer manager.push_ontology("another ontology") # no indication assert manager.get("cache_name", "key") is NotFound # check that the cache of the top level ontology is updated assert manager.current_cache_manager().caches["cache_name"].cache.copy() == {"key": Removed} def test_i_can_test_if_a_value_exists(self): sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache(extend_exists=lambda sdp, key: sdp.exists("cache_name", key))) manager.freeze() # default layer with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", "value") # add ontology layers manager.push_ontology("new ontology") manager.push_ontology("another ontology") # I can get the low level value assert manager.exists("cache_name", "key") # check that the value is not in cache (only in the low level database) assert not manager.current_cache_manager().has("cache_name", "key") assert not manager.ontologies[1].cache_manager.has("cache_name", "key") assert not manager.ontologies[2].cache_manager.has("cache_name", "key") def test_i_can_check_that_a_value_does_not_exist(self): sheerka = self.get_sheerka(cache_only=False) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache(extend_exists=lambda sdp, key: sdp.exists("cache_name", key))) manager.freeze() # add ontology layers manager.push_ontology("new ontology") manager.push_ontology("another ontology") assert not manager.exists("cache_name", "key") def test_i_can_list_from_multiple_ontologies(self): sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache()) manager.freeze() # default layer with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key1", DummyObj("key1", "value1")) manager.push_ontology("new ontology") with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key2", DummyObj("key2", "value2")) transaction.add("cache_name", "key1", DummyObj("key1", "value11")) # key1 is modified manager.push_ontology("another ontology") with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key3", DummyObj("key3", "value3")) assert manager.list("cache_name") == [DummyObj("key1", "value11"), DummyObj("key2", "value2"), DummyObj("key3", "value3")] def test_i_can_list_from_multiple_ontologies_even_if_they_are_not_all_filled(self): sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache()) manager.freeze() # default layer with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key1", DummyObj("key1", "value1")) manager.push_ontology("new ontology") # nothing in this ontology manager.push_ontology("another ontology") with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key3", DummyObj("key3", "value3")) assert manager.list("cache_name") == [DummyObj("key1", "value1"), DummyObj("key3", "value3")] def test_i_can_list_when_no_items(self): sheerka = self.get_sheerka(cache_only=False) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache()) manager.freeze() manager.push_ontology("new ontology") manager.push_ontology("another ontology") assert manager.list("cache_name") == [] def test_i_can_put_an_entry_cache_only_true(self): sheerka = self.get_sheerka(cache_only=True) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache()) manager.freeze() manager.put("cache_name", "key", "value") assert manager.get("cache_name", "key") == "value" manager.push_ontology("new ontology") manager.put("cache_name", "key", "value2") assert manager.get("cache_name", "key") == "value2" def test_i_can_put_entry_cache_only_false(self): sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache()) manager.freeze() # default ontology manager.put("cache_name", "key", "value") manager.commit(context) assert manager.get("cache_name", "key") == "value" assert manager.ontologies[0].cache_manager.sdp.state.data == {'cache_name': {'key': 'value'}} # add an ontology layer manager.push_ontology("new ontology") manager.put("cache_name", "key", "value2") manager.commit(context) assert manager.get("cache_name", "key") == "value2" assert manager.ontologies[0].cache_manager.sdp.state.data == {'cache_name': {'key': 'value2'}} assert manager.ontologies[1].cache_manager.sdp.state.data == {'cache_name': {'key': 'value'}} def test_i_can_put_in_a_list_cache(self): # in this test, sub layers have values. # We need to check that those values are not lost when adding a new element sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", ListCache().auto_configure("cache_name")) manager.freeze() # default ontology with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", ["val1"]) # add an ontology layer manager.push_ontology("new ontology") manager.put("cache_name", "key", "val2") manager.commit(context) assert manager.get("cache_name", "key") == ["val1", "val2"] assert manager.ontologies[0].cache_manager.sdp.state.data == {'cache_name': {'key': ['val1', 'val2']}} assert manager.ontologies[1].cache_manager.sdp.state.data == {'cache_name': {'key': ['val1']}} # and I can keep adding in another layer manager.push_ontology("another ontology") manager.put("cache_name", "key", "val3") manager.commit(context) assert manager.get("cache_name", "key") == ["val1", "val2", "val3"] assert manager.ontologies[0].cache_manager.sdp.state.data == {'cache_name': {'key': ['val1', 'val2', 'val3']}} assert manager.ontologies[1].cache_manager.sdp.state.data == {'cache_name': {'key': ['val1', 'val2']}} assert manager.ontologies[2].cache_manager.sdp.state.data == {'cache_name': {'key': ['val1']}} def test_i_can_remove_an_entry_that_is_only_in_db(self): sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache().auto_configure("cache_name")) manager.freeze() # default ontology # value in DB but not in cache with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", "value") manager.delete("cache_name", "key") manager.commit(context) assert manager.get("cache_name", "key") is NotFound # sanity check, the entry is removed assert manager.ontologies[0].cache_manager.caches["cache_name"].cache.copy() == {} assert manager.ontologies[0].cache_manager.sdp.state.data == {'cache_name': {}} def test_i_can_remove_when_value_is_in_low_level(self): # In this test, there is a value in a lower level ontology # After calling delete(), the value is no longer accessible, but not deleted sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache(default=lambda sdp, key: sdp.get("cache_name", key), extend_exists=lambda sdp, key: sdp.exists("cache_name", key))) manager.freeze() # default ontology # value in DB but not in cache with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", "value") # add an ontology layer manager.push_ontology("new ontology") manager.delete("cache_name", "key", "value") manager.commit(context) assert manager.get("cache_name", "key") is NotFound # sanity check, the entry is removed assert manager.ontologies[0].cache_manager.caches["cache_name"].cache.copy() == {"key": Removed} assert manager.ontologies[0].cache_manager.sdp.state.data == {'cache_name': {"key": Removed}} assert manager.ontologies[1].cache_manager.caches["cache_name"].cache.copy() == {} assert manager.ontologies[1].cache_manager.sdp.state.data == {'cache_name': {"key": "value"}} # The entry still exists in lower ontology manager.pop_ontology(context) assert manager.get("cache_name", "key") == "value" def test_i_can_remove_when_value_is_in_both_low_and_current_level(self): # In this test, there is a value is in a lower level ontology and in the current ontology # After calling delete(), the value is no longer accessible, but not deleted sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache(default=lambda sdp, key: sdp.get("cache_name", key), extend_exists=lambda sdp, key: sdp.exists("cache_name", key))) manager.freeze() # default ontology # value in DB but not in cache with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", "value") # add an ontology layer manager.push_ontology("new ontology") with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", "value2") manager.delete("cache_name", "key", "value") manager.commit(context) assert manager.get("cache_name", "key") is NotFound # sanity check, the entry is removed assert manager.ontologies[0].cache_manager.caches["cache_name"].cache.copy() == {"key": Removed} assert manager.ontologies[0].cache_manager.sdp.state.data == {'cache_name': {"key": Removed}} assert manager.ontologies[1].cache_manager.caches["cache_name"].cache.copy() == {} assert manager.ontologies[1].cache_manager.sdp.state.data == {'cache_name': {"key": "value"}} # The entry still exists in lower ontology manager.pop_ontology(context) assert manager.get("cache_name", "key") == "value" def test_i_can_remove_when_value_is_not_low_level(self): # In this test, there is a value is only in the current level # The value is deleted sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache().auto_configure("cache_name")) manager.freeze() # add an ontology layer # so that the value does not exists in the lower level ontology manager.push_ontology("new ontology") # value in DB but not in cache with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", "value") manager.delete("cache_name", "key") manager.commit(context) assert manager.get("cache_name", "key") is NotFound # sanity check, the entry is removed assert manager.ontologies[0].cache_manager.caches["cache_name"].cache.copy() == {} assert manager.ontologies[0].cache_manager.sdp.state.data == {'cache_name': {}} def test_i_can_remove_list_if_needed_when_value_is_in_low_level(self): # In this test, there are multiple values in a low level ontology # We remove only one, # So the top level ontology must be a copy minus the removed value sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) cache = ListIfNeededCache().auto_configure("cache_name") manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", cache) manager.freeze() # default ontology # value in DB but not in cache with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", ["value", "value2"]) # add ontology layers manager.push_ontology("new ontology") manager.push_ontology("another ontology") manager.delete("cache_name", "key", "value") manager.commit(context) assert manager.get("cache_name", "key") == "value2" # sanity check, the entry is removed assert manager.ontologies[0].cache_manager.caches["cache_name"].cache.copy() == {"key": "value2"} assert manager.ontologies[0].cache_manager.sdp.state.data == {'cache_name': {"key": "value2"}} assert manager.ontologies[1].cache_manager.caches["cache_name"].cache.copy() == {} assert manager.ontologies[1].cache_manager.sdp.state.data == {} assert manager.ontologies[2].cache_manager.caches["cache_name"].cache.copy() == {"key": ["value", "value2"]} assert manager.ontologies[2].cache_manager.sdp.state.data == {'cache_name': {"key": ["value", "value2"]}} # The entry still exists in lower ontology manager.pop_ontology(context) manager.pop_ontology(context) assert manager.get("cache_name", "key") == ["value", "value2"] def test_i_can_add_concept_default_layer(self): sheerka, context, foo = self.init_concepts("foo", cache_only=False) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) cache = Cache(default=lambda sdp, key: sdp.get("by_id", key), extend_exists=lambda sdp, key: sdp.get("by_id", key)) manager.register_concept_cache("by_id", cache, lambda obj: obj.id, use_ref=True) cache = ListIfNeededCache(default=lambda sdp, key: sdp.get("by_key", key), extend_exists=lambda sdp, key: sdp.get("by_key", key)) manager.register_concept_cache("by_key", cache, lambda obj: obj.key, use_ref=True) manager.freeze() manager.add_concept(foo) manager.commit(context) assert manager.get("by_key", foo.key) == foo assert manager.get("by_id", foo.id) == foo assert manager.current_sdp().get("by_key", foo.key) == foo assert manager.current_sdp().get("by_id", foo.id) == foo def test_i_can_add_concept_in_top_layer(self): sheerka, context, foo = self.init_concepts("foo", cache_only=False) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) cache = Cache(default=lambda sdp, key: sdp.get("by_id", key), extend_exists=lambda sdp, key: sdp.get("by_id", key)) manager.register_concept_cache("by_id", cache, lambda obj: obj.id, use_ref=True) cache = ListIfNeededCache(default=lambda sdp, key: sdp.get("by_key", key), extend_exists=lambda sdp, key: sdp.get("by_key", key)) manager.register_concept_cache("by_key", cache, lambda obj: obj.key, use_ref=True) manager.freeze() # add an ontology layer manager.push_ontology("new ontology") manager.add_concept(foo) manager.commit(context) assert manager.get("by_key", foo.key) == foo assert manager.get("by_id", foo.id) == foo assert manager.current_sdp().get("by_key", foo.key) == foo assert manager.current_sdp().get("by_id", foo.id) == foo # sanity check assert list(manager.ontologies[0].cache_manager.sdp.state.data.keys()) == ['by_id', 'by_key'] assert manager.ontologies[1].cache_manager.sdp.state.data == {} def test_i_can_update_concept_in_default_layer(self): sheerka, context, foo = self.init_concepts("foo", cache_only=False) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) cache = Cache(default=lambda sdp, key: sdp.get("by_id", key), extend_exists=lambda sdp, key: sdp.get("by_id", key)) manager.register_concept_cache("by_id", cache, lambda obj: obj.id, use_ref=True) cache = ListIfNeededCache(default=lambda sdp, key: sdp.get("by_key", key), extend_exists=lambda sdp, key: sdp.get("by_key", key)) manager.register_concept_cache("by_key", cache, lambda obj: obj.key, use_ref=True) manager.freeze() manager.add_concept(foo) modified = Concept().update_from(foo) modified.get_metadata().body = "new body" assert foo != modified manager.update_concept(foo, modified) manager.commit(context) assert manager.get("by_key", foo.key) == modified assert manager.get("by_id", foo.id) == modified def test_i_can_update_concept_in_top_layer(self): sheerka, context, foo = self.init_concepts("foo", cache_only=False) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) cache = Cache().auto_configure("by_id") manager.register_concept_cache("by_id", cache, lambda obj: obj.id, use_ref=True) cache = ListIfNeededCache().auto_configure("by_key") manager.register_concept_cache("by_key", cache, lambda obj: obj.key, use_ref=True) manager.freeze() manager.add_concept(foo) manager.commit(context) # add an ontology layer manager.push_ontology("new ontology") modified = Concept().update_from(foo) modified.get_metadata().body = "new body" assert foo != modified manager.update_concept(foo, modified) manager.commit(context) assert manager.get("by_key", foo.key) == modified assert manager.get("by_id", foo.id) == modified # sanity check. # make sure that the previous values are kept assert manager.ontologies[0].cache_manager.sdp.get('by_key', foo.key) == modified assert manager.ontologies[0].cache_manager.sdp.get('by_id', foo.id) == modified assert manager.ontologies[1].cache_manager.sdp.get('by_key', foo.key) == foo assert manager.ontologies[1].cache_manager.sdp.get('by_id', foo.id) == foo # so I can get the old values when I pop ontology manager.pop_ontology(context) assert manager.get("by_key", foo.key) == foo assert manager.get("by_id", foo.id) == foo def test_i_can_update_when_concept_in_both_top_and_bottom_layers(self): sheerka, context, foo = self.init_concepts("foo", cache_only=False) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) cache = Cache(default=lambda sdp, key: sdp.get("by_id", key), extend_exists=lambda sdp, key: sdp.get("by_id", key)) manager.register_concept_cache("by_id", cache, lambda obj: obj.id, use_ref=True) cache = ListIfNeededCache(default=lambda sdp, key: sdp.get("by_key", key), extend_exists=lambda sdp, key: sdp.get("by_key", key)) manager.register_concept_cache("by_key", cache, lambda obj: obj.key, use_ref=True) manager.freeze() manager.add_concept(foo) manager.commit(context) # add an ontology layer # and modify the concept # The database is updated, but not the internal cache manager.push_ontology("new ontology") modified1 = Concept().update_from(foo) modified1.get_metadata().body = "new body" assert foo != modified1 with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("by_key", foo.key, modified1) transaction.add("by_id", foo.id, modified1) # modify the top layer a second time modified2 = Concept().update_from(foo) modified2.get_metadata().pre = "True" manager.update_concept(foo, modified2) manager.commit(context) assert manager.get("by_key", foo.key) == modified2 assert manager.get("by_id", foo.id) == modified2 # sanity check. # make sure that the previous values are kept # sanity check assert manager.ontologies[0].cache_manager.sdp.get('by_key', foo.key) == modified2 assert manager.ontologies[0].cache_manager.sdp.get('by_id', foo.id) == modified2 assert manager.ontologies[1].cache_manager.sdp.get('by_key', foo.key) == foo assert manager.ontologies[1].cache_manager.sdp.get('by_id', foo.id) == foo # so I can get the old values when I pop ontology manager.pop_ontology(context) assert manager.get("by_key", foo.key) == foo assert manager.get("by_id", foo.id) == foo def test_i_can_update_when_the_key_changes(self): sheerka, context, foo = self.init_concepts("foo", cache_only=False) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) cache = Cache().auto_configure("by_id") manager.register_concept_cache("by_id", cache, lambda obj: obj.id, use_ref=True) cache = ListIfNeededCache().auto_configure("by_key") manager.register_concept_cache("by_key", cache, lambda obj: obj.key, use_ref=True) manager.freeze() # create an entry manager.add_concept(foo) manager.commit(context) # add a new layer, and modify the concept manager.push_ontology("new ontology") modified = Concept().update_from(foo) modified.get_metadata().key = "another key" manager.update_concept(foo, modified) manager.commit(context) assert manager.get("by_id", modified.id) == modified assert manager.get("by_key", modified.key) == modified assert manager.get("by_key", foo.key) == NotFound # sanity assert manager.ontologies[0].cache_manager.sdp.get('by_key', foo.key) == Removed assert manager.ontologies[0].cache_manager.sdp.get('by_key', modified.key) == modified assert manager.ontologies[0].cache_manager.sdp.get('by_id', foo.id) == modified assert manager.ontologies[1].cache_manager.sdp.get('by_key', foo.key) == foo assert manager.ontologies[1].cache_manager.sdp.get('by_key', modified.key) == NotFound assert manager.ontologies[1].cache_manager.sdp.get('by_id', foo.id) == foo def test_i_can_update_when_key_changes_and_there_are_lists(self): sheerka, context, foo, foo2, bar = self.init_concepts("foo", Concept("foo", body="x"), "bar", cache_only=False) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) cache = Cache().auto_configure("by_id") manager.register_concept_cache("by_id", cache, lambda obj: obj.id, use_ref=True) cache = ListIfNeededCache().auto_configure("by_key") manager.register_concept_cache("by_key", cache, lambda obj: obj.key, use_ref=True) manager.freeze() # create entries manager.add_concept(foo) manager.add_concept(foo2) manager.add_concept(bar) manager.commit(context) # add a new layer, and modify the concept manager.push_ontology("new ontology") modified = Concept().update_from(foo) modified.get_metadata().key = "bar" manager.update_concept(foo, modified) manager.commit(context) assert manager.get("by_id", modified.id) == modified assert manager.get("by_key", modified.key) == [bar, modified] assert manager.get("by_key", foo.key) == foo2 # sanity check assert manager.ontologies[0].cache_manager.sdp.get('by_key', foo.key) == foo2 assert manager.ontologies[0].cache_manager.sdp.get('by_key', modified.key) == [bar, modified] assert manager.ontologies[0].cache_manager.sdp.get('by_id', foo.id) == modified assert manager.ontologies[1].cache_manager.sdp.get('by_key', foo.key) == [foo, foo2] assert manager.ontologies[1].cache_manager.sdp.get('by_key', modified.key) == bar assert manager.ontologies[1].cache_manager.sdp.get('by_id', foo.id) == foo def test_i_can_remove_concept_from_default_layer(self): sheerka, context, foo = self.init_concepts("foo", cache_only=False) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) cache = Cache(default=lambda sdp, key: sdp.get("by_id", key), extend_exists=lambda sdp, key: sdp.get("by_id", key)) manager.register_concept_cache("by_id", cache, lambda obj: obj.id, use_ref=True) cache = ListIfNeededCache(default=lambda sdp, key: sdp.get("by_key", key), extend_exists=lambda sdp, key: sdp.get("by_key", key)) manager.register_concept_cache("by_key", cache, lambda obj: obj.key, use_ref=True) manager.freeze() manager.add_concept(foo) manager.commit(context) manager.remove_concept(foo) manager.commit(context) assert manager.get("by_id", foo.id) == NotFound assert manager.get("by_key", foo.key) == NotFound # sanity check assert manager.current_sdp().get("by_key") == {} assert manager.current_sdp().get("by_id") == {} def test_i_can_remove_concept_from_top_layer(self): sheerka, context, foo = self.init_concepts("foo", cache_only=False) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) cache = Cache().auto_configure("by_id") manager.register_concept_cache("by_id", cache, lambda obj: obj.id, use_ref=True) cache = ListIfNeededCache().auto_configure("by_key") manager.register_concept_cache("by_key", cache, lambda obj: obj.key, use_ref=True) manager.freeze() manager.add_concept(foo) manager.commit(context) # add a new layer, and remove the concept manager.push_ontology("new ontology") manager.push_ontology("another ontology") manager.remove_concept(foo) manager.commit(context) assert manager.get("by_id", foo.id) == NotFound assert manager.get("by_key", foo.key) == NotFound # sanity check assert manager.current_sdp().get("by_id") == {foo.id: Removed} assert manager.current_sdp().get("by_key") == {foo.key: Removed} assert manager.ontologies[1].cache_manager.sdp.get("by_id") == NotFound assert manager.ontologies[1].cache_manager.sdp.get("by_key") == NotFound assert manager.ontologies[2].cache_manager.sdp.get("by_id") == {foo.id: foo} assert manager.ontologies[2].cache_manager.sdp.get("by_key") == {foo.key: foo} # So I can pop manager.pop_ontology(context) assert manager.get("by_id", foo.id) == foo assert manager.get("by_key", foo.key) == foo # and pop again manager.pop_ontology(context) assert manager.get("by_id", foo.id) == foo assert manager.get("by_key", foo.key) == foo def test_i_can_get_all(self): sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache().auto_configure("cache_name")) manager.freeze() manager.put("cache_name", "key_to_remove1", "value1") manager.put("cache_name", "key_to_remove2", "value1") manager.put("cache_name", "key1", "value1") manager.put("cache_name", "key2", "value2_in_sdp") manager.put("cache_name", "key3", "value3") manager.commit(context) manager.put("cache_name", "key2", "value2_in_cache") # in cache, but not in remote sdp # add ontology layers manager.push_ontology("new ontology") manager.put("cache_name", "key1", "value1_from_new_ontology") manager.put("cache_name", "key2", "value2_from_new_ontology") manager.put("cache_name", "key4", "value4_in_sdp") manager.commit(context) manager.put("cache_name", "key4", "value4_in_cache") manager.put("cache_name", "key_to_remove1", Removed) manager.push_ontology("another ontology") with manager.current_sdp().get_transaction(context.event) as transaction: # so that value is only in sdp, not in cache transaction.add("cache_name", "key5", "value5") transaction.add("cache_name", "key_to_remove2", Removed) assert manager.get_all("cache_name") == { "key1": "value1_from_new_ontology", "key2": "value2_from_new_ontology", "key3": "value3", "key4": "value4_in_cache", "key5": "value5" } def test_i_can_keep_track_of_created_concepts_by_ontologies(self): sheerka, context, foo = self.init_concepts("foo", create_new=True) def from_cache(entry): return sheerka.om.self_cache_manager.copy(entry) def from_db(entry): return sheerka.om.self_cache_manager.sdp.get(entry) # check that the new concept is tracked assert from_cache(SheerkaOntologyManager.CONCEPTS_BY_ONTOLOGY_ENTRY) == {'#unit_test#': {'1001'}} assert from_cache(SheerkaOntologyManager.ONTOLOGY_BY_CONCEPT_ENTRY) == {'1001': '#unit_test#'} # add a new ontology and make sure the new concepts are tracked sheerka.push_ontology(context, "new ontology") sheerka.create_new_concept(context, Concept("bar")) assert from_cache(SheerkaOntologyManager.CONCEPTS_BY_ONTOLOGY_ENTRY) == {'#unit_test#': {'1001'}, 'new ontology': {'1002'}} assert from_cache(SheerkaOntologyManager.ONTOLOGY_BY_CONCEPT_ENTRY) == {'1001': '#unit_test#', '1002': 'new ontology'} # commit the info and check the DB sheerka.om.commit(context) assert from_db(SheerkaOntologyManager.CONCEPTS_BY_ONTOLOGY_ENTRY) == {'#unit_test#': {'1001'}, 'new ontology': {'1002'}, } assert from_db(SheerkaOntologyManager.ONTOLOGY_BY_CONCEPT_ENTRY) == {'1001': '#unit_test#', '1002': 'new ontology', } # remove a concept a check sheerka.remove_concept(context, sheerka.get_by_name("foo")) assert from_cache(SheerkaOntologyManager.CONCEPTS_BY_ONTOLOGY_ENTRY) == {'new ontology': {'1002'}, } assert from_cache(SheerkaOntologyManager.ONTOLOGY_BY_CONCEPT_ENTRY) == {'1002': 'new ontology', } sheerka.remove_concept(context, sheerka.get_by_name("bar")) assert from_cache(SheerkaOntologyManager.CONCEPTS_BY_ONTOLOGY_ENTRY) == {} assert from_cache(SheerkaOntologyManager.ONTOLOGY_BY_CONCEPT_ENTRY) == {} # commit again and check sheerka.om.commit(context) assert from_db(SheerkaOntologyManager.CONCEPTS_BY_ONTOLOGY_ENTRY) == {} assert from_db(SheerkaOntologyManager.ONTOLOGY_BY_CONCEPT_ENTRY) == {} def test_i_can_keep_track_of_created_rules_by_ontologies(self): sheerka, context, rule1 = self.init_format_rules(("rule1", "id.attr == 'value'", "True")) def rules_by_ontology_from_cache(): res = sheerka.om.self_cache_manager.copy(SheerkaOntologyManager.RULES_BY_ONTOLOGY_ENTRY) del res[SheerkaOntologyManager.ROOT_ONTOLOGY_NAME] # discard builtin rules return res def ontologies_from_cache(): res = sheerka.om.self_cache_manager.copy(SheerkaOntologyManager.ONTOLOGY_BY_RULE_ENTRY) return {k: v for k, v in res.items() if v != SheerkaOntologyManager.ROOT_ONTOLOGY_NAME} def rules_by_ontology_from_db(): res = sheerka.om.self_cache_manager.sdp.get(SheerkaOntologyManager.RULES_BY_ONTOLOGY_ENTRY) del res[SheerkaOntologyManager.ROOT_ONTOLOGY_NAME] # discard builtin rules return res def ontologies_from_db(): res = sheerka.om.self_cache_manager.sdp.get(SheerkaOntologyManager.ONTOLOGY_BY_RULE_ENTRY) return {k: v for k, v in res.items() if v != SheerkaOntologyManager.ROOT_ONTOLOGY_NAME} assert rules_by_ontology_from_cache() == {"#unit_test#": {rule1.id}} assert ontologies_from_cache() == {rule1.id: "#unit_test#"} # add a new rule from a new ontology and check sheerka.push_ontology(context, "new ontology") rule2 = Rule(ACTION_TYPE_EXEC, "rule2", "id2.attr2 == 'value'", "True") sheerka.create_new_rule(context, rule2) assert rules_by_ontology_from_cache() == {"#unit_test#": {rule1.id}, "new ontology": {rule2.id}} assert ontologies_from_cache() == {rule1.id: "#unit_test#", rule2.id: "new ontology"} # commit and check the result sheerka.om.commit(context) assert rules_by_ontology_from_db() == {"#unit_test#": {rule1.id}, "new ontology": {rule2.id}} assert ontologies_from_db() == {rule1.id: "#unit_test#", rule2.id: "new ontology"} sheerka.remove_rule(context, rule1) assert rules_by_ontology_from_cache() == {"new ontology": {rule2.id}} assert ontologies_from_cache() == {rule2.id: "new ontology"} # remove the last rule sheerka.remove_rule(context, rule2) assert rules_by_ontology_from_cache() == {} assert ontologies_from_cache() == {} # commit and check the db sheerka.om.commit(context) assert rules_by_ontology_from_db() == {} assert ontologies_from_db() == {} def test_i_can_keep_track_of_created_concept_on_ontology_pop(self): sheerka, context, foo = self.init_concepts("foo", create_new=True) events_raised = set() sheerka.subscribe(EVENT_CONCEPT_ID_DELETED, lambda ctx, c: events_raised.add(c)) def from_cache(entry): return sheerka.om.self_cache_manager.copy(entry) sheerka.push_ontology(context, "new ontology") sheerka.create_new_concept(context, Concept("bar")) sheerka.create_new_concept(context, Concept("baz")) sheerka.pop_ontology(context) assert from_cache(SheerkaOntologyManager.CONCEPTS_BY_ONTOLOGY_ENTRY) == {'#unit_test#': {'1001'}} assert from_cache(SheerkaOntologyManager.ONTOLOGY_BY_CONCEPT_ENTRY) == {'1001': '#unit_test#'} # check that the 'concept is deleted' events are raised assert events_raised == {'1002', '1003'} def test_i_can_keep_track_of_created_rules_on_ontology_pop(self): sheerka, context, rule1 = self.init_format_rules(("rule1", "id.attr == 'value'", "True")) events_raised = set() sheerka.subscribe(EVENT_RULE_ID_DELETED, lambda ctx, r: events_raised.add(r)) def rules_by_ontology_from_cache(): res = sheerka.om.self_cache_manager.copy(SheerkaOntologyManager.RULES_BY_ONTOLOGY_ENTRY) del res[SheerkaOntologyManager.ROOT_ONTOLOGY_NAME] # discard builtin rules return res def ontologies_from_cache(): res = sheerka.om.self_cache_manager.copy(SheerkaOntologyManager.ONTOLOGY_BY_RULE_ENTRY) return {k: v for k, v in res.items() if v != SheerkaOntologyManager.ROOT_ONTOLOGY_NAME} sheerka.push_ontology(context, "new ontology") sheerka.create_new_rule(context, Rule(ACTION_TYPE_EXEC, "rule2", "id2.attr2 == 'value'", "True")) sheerka.create_new_rule(context, Rule(ACTION_TYPE_EXEC, "rule3", "id3.attr3 == 'value'", "True")) sheerka.pop_ontology(context) assert rules_by_ontology_from_cache() == {'#unit_test#': {'10'}} assert ontologies_from_cache() == {'10': '#unit_test#'} # check that the 'rule is deleted' events are raised assert events_raised == {'11', '12'} # def test_i_can_list_by_key_when_dictionaries(self): # sheerka = self.get_sheerka(cache_only=False) # context = self.get_context(sheerka) # # manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) # manager.register_cache("cache_name", Cache().auto_configure("cache_name")) # manager.freeze() # # manager.put("cache_name", "key1", {"a": "value1", "b": "value2", "c": "value3"}) # manager.commit(context) # # manager.push_ontology("new ontology") # manager.put("cache_name", "key1", {"a": "new value1", "d": "value4"}) # only in cache # # manager.push_ontology("another ontology") # with manager.current_sdp().get_transaction(context.event) as transaction: # transaction.add("cache_name", "key1", {"b": "new value2", "e": "value5"}) # # assert manager.list_by_key("cache_name", "key1") == { # "a": "new value1", # "b": "new value2", # "c": "value3", # "d": "value4", # "e": "value5", # } # # def test_i_can_list_by_key_when_lists(self): # sheerka = self.get_sheerka(cache_only=False) # context = self.get_context(sheerka) # # manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) # manager.register_cache("cache_name", Cache().auto_configure("cache_name")) # manager.freeze() # # manager.put("cache_name", "key1", ["a", "b", "c"]) # manager.commit(context) # # manager.push_ontology("new ontology") # manager.put("cache_name", "key1", ["a", "d"]) # only in cache # # manager.push_ontology("another ontology") # with manager.current_sdp().get_transaction(context.event) as transaction: # transaction.add("cache_name", "key1", ["b", "e"]) # # assert manager.list_by_key("cache_name", "key1") == ["a", "b", "c", "a", "d", "b", "e"] # # def test_i_can_list_by_key_when_dictionaries_and_entries_are_removed(self): # sheerka = self.get_sheerka(cache_only=False) # context = self.get_context(sheerka) # # manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) # manager.register_cache("cache_name", Cache().auto_configure("cache_name")) # manager.freeze() # # manager.put("cache_name", "key1", {"a": "value1", "b": "value2", "c": "value3"}) # manager.put("cache_name", "key2", {"a": "value1", "b": "value2", "c": "value3"}) # manager.put("cache_name", "key3", {"a": "value1", "b": "value2", "c": "value3"}) # manager.commit(context) # # manager.push_ontology("new ontology") # manager.put("cache_name", "key1", Removed) # removed in cache # with manager.current_sdp().get_transaction(context.event) as transaction: # transaction.add("cache_name", "key2", Removed) # removed in sdp # # manager.push_ontology("another ontology") # manager.put("cache_name", "key1", {"e": "value1", "f": "value2", "g": "value3"}) # manager.put("cache_name", "key2", {"e": "value1", "f": "value2", "g": "value3"}) # manager.put("cache_name", "key3", {"e": "value1", "f": "value2", "g": "value3"}) # # assert manager.list_by_key("cache_name", "key1") == {"e": "value1", "f": "value2", "g": "value3"} # assert manager.list_by_key("cache_name", "key2") == {"e": "value1", "f": "value2", "g": "value3"} # assert manager.list_by_key("cache_name", "key3") == {"a": "value1", "b": "value2", "c": "value3", # "e": "value1", "f": "value2", "g": "value3"} # # def test_i_can_list_by_key_when_lists_and_entries_are_removed(self): # sheerka = self.get_sheerka(cache_only=False) # context = self.get_context(sheerka) # # manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) # manager.register_cache("cache_name", Cache().auto_configure("cache_name")) # manager.freeze() # # manager.put("cache_name", "key1", ["a", "b", "c"]) # manager.put("cache_name", "key2", ["a", "b", "c"]) # manager.put("cache_name", "key3", ["a", "b", "c"]) # manager.commit(context) # # manager.push_ontology("new ontology") # manager.put("cache_name", "key1", Removed) # removed in cache # with manager.current_sdp().get_transaction(context.event) as transaction: # transaction.add("cache_name", "key2", Removed) # removed in sdp # # manager.push_ontology("another ontology") # manager.put("cache_name", "key1", ["e", "f", "g"]) # manager.put("cache_name", "key2", ["e", "f", "g"]) # manager.put("cache_name", "key3", ["e", "f", "g"]) # # assert manager.list_by_key("cache_name", "key1") == ["e", "f", "g"] # assert manager.list_by_key("cache_name", "key2") == ["e", "f", "g"] # assert manager.list_by_key("cache_name", "key3") == ["a", "b", "c", "e", "f", "g"] def test_i_can_get_call_when_a_cache_is_cleared(self): sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache().auto_configure("cache_name")) manager.freeze() manager.put("cache_name", "key1", "value1") manager.put("cache_name", "key2", "value2") manager.put("cache_name", "key3", "value3") # add ontology layers manager.push_ontology("new ontology") manager.clear("cache_name") manager.put("cache_name", "key1", "new value1") manager.put("cache_name", "key4", "value4") manager.push_ontology("another ontology") manager.put("cache_name", "key5", "value5") assert manager.get_all("cache_name") == { "key1": "new value1", "key4": "value4", "key5": "value5" } def test_i_can_get_all_when_inc_cache(self): sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", IncCache().auto_configure("cache_name")) manager.freeze() assert manager.get("cache_name", "key1") == 1 assert manager.get("cache_name", "key1") == 2 manager.push_ontology("new ontology") assert manager.get("cache_name", "key1") == 3 assert manager.get("cache_name", "key2") == 1 assert manager.get("cache_name", "key2") == 2 assert manager.get_all("cache_name") == { "key1": 3, "key2": 2, } # a second time, to make sure that nothing was incremented assert manager.get_all("cache_name") == { "key1": 3, "key2": 2, } @pytest.mark.parametrize("all_ontologies, expected_in_layer_1", [ (False, {}), (True, {'key1': DummyObj(key='key1', value='value1'), 'key2': DummyObj(key='key2', value='value2')}), ]) def test_i_can_populate(self, all_ontologies, expected_in_layer_1): sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache().auto_configure("cache_name")) manager.freeze() manager.put("cache_name", "key1", DummyObj("key1", "value1")) manager.put("cache_name", "key2", DummyObj("key2", "value2")) manager.commit(context) manager.clear("cache_name") manager.push_ontology("new ontology") manager.put("cache_name", "key2", DummyObj("key2", "value22")) manager.put("cache_name", "key3", DummyObj("key3", "value3")) manager.commit(context) manager.clear("cache_name") # sanity check assert manager.ontologies[0].cache_manager.get_cache("cache_name").copy() == {} assert manager.ontologies[1].cache_manager.get_cache("cache_name").copy() == {} manager.populate("cache_name", lambda sdp: sdp.list("cache_name"), lambda obj: obj.key, all_ontologies=all_ontologies) assert manager.ontologies[0].cache_manager.get_cache("cache_name").copy() == { 'key2': DummyObj(key='key2', value='value22'), 'key3': DummyObj(key='key3', value='value3')} assert manager.ontologies[1].cache_manager.get_cache("cache_name").copy() == expected_in_layer_1 def test_i_can_clear_when_multiple_ontology_layers(self): sheerka = self.get_sheerka(cache_only=False) context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache().auto_configure("cache_name")) manager.freeze() manager.put("cache_name", "key1", "value1") manager.put("cache_name", "key2", "value2") assert manager.ontologies[0].cache_manager.get_cache("cache_name").copy() == {'key1': 'value1', 'key2': 'value2'} # I can clear in another layer manager.push_ontology("new ontology") manager.clear("cache_name") assert manager.get("cache_name", "key1") is NotFound assert manager.get("cache_name", "key2") is NotFound assert manager.ontologies[0].cache_manager.get_cache("cache_name").copy() == {} manager.put("cache_name", "key1", "new value1") assert manager.get("cache_name", "key1") == "new value1" assert manager.get("cache_name", "key2") is NotFound assert manager.ontologies[0].cache_manager.get_cache("cache_name").copy() == {'key1': "new value1"} manager.push_ontology("another ontology") manager.put("cache_name", "key2", "new value2") assert manager.get("cache_name", "key1") == "new value1" assert manager.get("cache_name", "key2") == "new value2" assert manager.ontologies[0].cache_manager.get_cache("cache_name").copy() == {'key1': "new value1", 'key2': "new value2"} manager.clear("cache_name") assert manager.get("cache_name", "key1") is NotFound assert manager.get("cache_name", "key2") is NotFound assert manager.ontologies[0].cache_manager.get_cache("cache_name").copy() == {} assert manager.ontologies[1].cache_manager.get_cache("cache_name").copy() == {'key1': "new value1"} assert manager.ontologies[2].cache_manager.get_cache("cache_name").copy() == {'key1': 'value1', 'key2': 'value2'} manager.pop_ontology(context) assert manager.get("cache_name", "key1") == "new value1" assert manager.get("cache_name", "key2") is NotFound manager.pop_ontology(context) assert manager.get("cache_name", "key1") == "value1" assert manager.get("cache_name", "key2") == "value2" def test_already_on_the_top(self): sheerka = self.get_sheerka() manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.freeze() manager.push_ontology("new ontology") assert manager.already_on_top("new ontology") assert not manager.already_on_top("another ontology") def test_already_on_the_top_when_the_ontology_already_exists(self): sheerka = self.get_sheerka() manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.freeze() manager.push_ontology("new ontology") manager.push_ontology("another ontology") with pytest.raises(OntologyAlreadyExists): assert manager.already_on_top("new ontology") class TestSheerkaOntologyWithFileBasedSheerka(TestUsingFileBasedSheerka): def test_i_can_put_back_ontology(self): sheerka = self.get_sheerka() context = self.get_context(sheerka) manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only) manager.register_cache("cache_name", Cache().auto_configure("cache_name")) manager.freeze() # default layer with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", "value1") # add a layer manager.push_ontology("new ontology") with manager.current_sdp().get_transaction(context.event) as transaction: transaction.add("cache_name", "key", "value2") assert manager.get("cache_name", "key") == "value2" manager.pop_ontology(context) assert manager.get("cache_name", "key") == "value1" # put back the previous ontology manager.push_ontology("new ontology") assert manager.get("cache_name", "key") == "value2" def test_i_can_remember_concept_and_rules_by_ontology(self): sheerka, context, foo, r1 = self.init_test().with_concepts( "foo", create_new=True ).with_format_rules( ("rule1", "__ret", "True"), ).unpack() sheerka.om.commit(context) sheerka = self.new_sheerka_instance(False) context = self.get_context(sheerka) sheerka.create_new_concept(context, Concept("bar")) r2 = sheerka.create_new_rule(context, Rule(ACTION_TYPE_EXEC, "rule2", "__ret.status", "True")).body.body sheerka.om.commit(context) sheerka.push_ontology(context, "new ontology") sheerka.create_new_concept(context, Concept("baz")) sheerka.create_new_rule(context, Rule(ACTION_TYPE_EXEC, "rule3", "id3.attr3 == 'value'", "True")) sheerka.om.commit(context) sheerka = self.new_sheerka_instance(False) context = self.get_context(sheerka) sheerka.push_ontology(context, "another ontology") sheerka.create_new_concept(context, Concept("qux")) r4 = sheerka.create_new_rule(context, Rule(ACTION_TYPE_EXEC, "rule4", "id4.attr4", "True")).body.body sheerka.remove_concept(context, foo) sheerka.remove_rule(context, r2) sheerka.om.commit(context) assert sheerka.om.self_cache_manager.copy(SheerkaOntologyManager.CONCEPTS_BY_ONTOLOGY_ENTRY) == { '#unit_test#': {'1002'}, 'another ontology': {'1004'}, } assert sheerka.om.self_cache_manager.copy(SheerkaOntologyManager.RULES_BY_ONTOLOGY_ENTRY) == { '#unit_test#': {r1.id}, 'another ontology': {r4.id}, } # in db assert sheerka.om.self_cache_manager.sdp.get(SheerkaOntologyManager.CONCEPTS_BY_ONTOLOGY_ENTRY) == { '#unit_test#': {'1002'}, 'another ontology': {'1004'}, 'new ontology': {'1003'}} rules_from_db = sheerka.om.self_cache_manager.sdp.get(SheerkaOntologyManager.RULES_BY_ONTOLOGY_ENTRY) del rules_from_db["__default__"] assert rules_from_db == { '#unit_test#': {'10'}, 'another ontology': {'13'}, 'new ontology': {'12'}}