kodjo
87cab44fb8
Fixed #135: Change services service priorities Fixed #136: ErrorManager: Implement recognize_error Fixed #137: BNFNodeParser : Error when parsing regex with sub parsers Fixed #138: get_last_errors(): real errors sources are lost Fixed #139: OneError return value removes the origin of the error Fixed #140: Concept variables are not correctly handled when parsing sub expression Fixed #143: Implement has_unknown_concepts()
1670 lines
75 KiB
Python
1670 lines
75 KiB
Python
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().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)
|
|
|
|
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().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()
|
|
|
|
# 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_add_the_concepts_with_the_same_key_from_different_layers(self):
|
|
sheerka, context, foo_1, foo_2 = self.init_concepts(
|
|
Concept("foo x", body="x + 1").def_var("x"),
|
|
Concept("foo x", body="x + 2").def_var("x"),
|
|
cache_only=False)
|
|
|
|
manager = SheerkaOntologyManager(sheerka, sheerka.root_folder, sheerka.cache_only)
|
|
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_1)
|
|
manager.commit(context)
|
|
|
|
manager.push_ontology("new ontology")
|
|
manager.add_concept(foo_2)
|
|
manager.commit(context)
|
|
|
|
assert manager.current_sdp().get("by_key", foo_1.key) == [foo_1, foo_2]
|
|
|
|
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().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)
|
|
|
|
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().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
|
|
# 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#': {'11'}}
|
|
assert ontologies_from_cache() == {'11': '#unit_test#'}
|
|
|
|
# check that the 'rule is deleted' events are raised
|
|
assert events_raised == {'12', '13'}
|
|
|
|
# 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#': {'11'},
|
|
'another ontology': {'14'},
|
|
'new ontology': {'13'}}
|