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()
303 lines
13 KiB
Python
303 lines
13 KiB
Python
from core.builtin_concepts import BuiltinConcepts
|
|
from core.builtin_helpers import ensure_evaluated
|
|
from core.concept import Concept
|
|
from core.global_symbols import NotFound
|
|
from core.sheerka.ExecutionContext import ExecutionContext
|
|
from core.sheerka.services.SheerkaMemory import MemoryObject, SheerkaMemory
|
|
from tests.TestUsingFileBasedSheerka import TestUsingFileBasedSheerka
|
|
from tests.TestUsingMemoryBasedSheerka import TestUsingMemoryBasedSheerka
|
|
from tests.parsers.parsers_utils import CB, CV, compare_with_test_object
|
|
|
|
|
|
class TestSheerkaMemory(TestUsingMemoryBasedSheerka):
|
|
def test_i_can_add_to_global_short_term_memory(self):
|
|
sheerka = self.get_sheerka()
|
|
service = sheerka.services[SheerkaMemory.NAME]
|
|
|
|
foo = Concept("foo")
|
|
sheerka.add_to_short_term_memory(None, "a", foo)
|
|
|
|
assert service.short_term_objects.copy() == {'global': {'a': foo}}
|
|
assert id(sheerka.get_from_short_term_memory(None, "a")) == id(foo)
|
|
|
|
def test_i_can_add_context_short_term_memory(self):
|
|
sheerka, context = self.init_test().unpack()
|
|
service = sheerka.services[SheerkaMemory.NAME]
|
|
|
|
foo = Concept("foo")
|
|
sheerka.add_to_short_term_memory(context, "a", foo)
|
|
|
|
context_id = ExecutionContext.ids[context.event.get_digest()]
|
|
assert service.short_term_objects.copy() == {context_id: {'a': foo}}
|
|
assert id(sheerka.get_from_short_term_memory(context, "a")) == id(foo)
|
|
assert sheerka.get_from_short_term_memory(None, "a") is NotFound
|
|
|
|
def test_i_can_add_many_to_short_term_memory(self):
|
|
sheerka, context = self.init_test().unpack()
|
|
bag = {"a": "foo", "b": "bar", }
|
|
context_id = ExecutionContext.ids[context.event.get_digest()]
|
|
service = sheerka.services[SheerkaMemory.NAME]
|
|
|
|
sheerka.add_many_to_short_term_memory(context, bag)
|
|
assert service.short_term_objects.copy() == {context_id: bag}
|
|
|
|
def test_i_can_get_obj_from_parents(self):
|
|
sheerka, context = self.init_test().unpack()
|
|
foo = Concept("foo")
|
|
|
|
sheerka.add_to_short_term_memory(None, "a", foo)
|
|
sheerka.add_to_short_term_memory(context, "b", foo)
|
|
|
|
with context.push(BuiltinConcepts.TESTING, None) as sub_context:
|
|
assert id(sheerka.get_from_short_term_memory(sub_context, "a")) == id(foo)
|
|
assert id(sheerka.get_from_short_term_memory(context, "a")) == id(foo)
|
|
assert id(sheerka.get_from_short_term_memory(None, "a")) == id(foo)
|
|
|
|
assert id(sheerka.get_from_short_term_memory(sub_context, "b")) == id(foo)
|
|
assert id(sheerka.get_from_short_term_memory(context, "b")) == id(foo)
|
|
assert sheerka.get_from_short_term_memory(None, "b") is NotFound
|
|
|
|
def test_short_term_memory_entries_are_removed_on_context_exit(self):
|
|
sheerka, context = self.init_test().unpack()
|
|
|
|
with context.push(BuiltinConcepts.TESTING, None) as sub_context:
|
|
foo = Concept("foo")
|
|
sheerka.add_to_short_term_memory(sub_context, "a", foo)
|
|
assert id(sheerka.get_from_short_term_memory(sub_context, "a")) == id(foo)
|
|
|
|
assert sheerka.get_from_short_term_memory(sub_context, "a") is NotFound
|
|
|
|
def test_short_term_memory_entries_are_removed_on_context_exit_2(self):
|
|
# this time we test the bulk insert
|
|
sheerka, context = self.init_test().unpack()
|
|
|
|
with context.push(BuiltinConcepts.TESTING, None) as sub_context:
|
|
foo = Concept("foo")
|
|
sheerka.add_many_to_short_term_memory(sub_context, {"a": foo})
|
|
assert id(sheerka.get_from_short_term_memory(sub_context, "a")) == id(foo)
|
|
|
|
assert sheerka.get_from_short_term_memory(sub_context, "a") is NotFound
|
|
|
|
def test_i_can_add_and_retrieve_from_memory(self):
|
|
sheerka, context = self.init_test().unpack()
|
|
|
|
assert sheerka.get_from_memory(context, "a") is NotFound
|
|
|
|
foo = Concept("foo")
|
|
sheerka.add_to_memory(context, "a", foo)
|
|
|
|
assert sheerka.om.copy(SheerkaMemory.OBJECTS_ENTRY) == {"a": MemoryObject(context.event.get_digest(),
|
|
context.event.date.timestamp(),
|
|
foo)}
|
|
assert id(sheerka.get_from_memory(context, "a").obj) == id(foo)
|
|
|
|
# another object is appended
|
|
bar = Concept("bar")
|
|
sheerka.add_to_memory(context, "a", bar)
|
|
assert sheerka.om.copy(SheerkaMemory.OBJECTS_ENTRY) == {
|
|
"a": [
|
|
MemoryObject(context.event.get_digest(), context.event.date.timestamp(), foo),
|
|
MemoryObject(context.event.get_digest(), context.event.date.timestamp(), bar)
|
|
]}
|
|
|
|
def test_i_can_set_and_retrieve_from_memory(self):
|
|
sheerka, context = self.init_test().unpack()
|
|
|
|
assert sheerka.get_from_memory(context, "a") is NotFound
|
|
|
|
foo = Concept("foo")
|
|
sheerka.set_in_memory(context, "a", foo)
|
|
assert sheerka.om.copy(SheerkaMemory.OBJECTS_ENTRY) == {"a": MemoryObject(context.event.get_digest(),
|
|
context.event.date.timestamp(),
|
|
foo)}
|
|
assert id(sheerka.get_from_memory(context, "a").obj) == id(foo)
|
|
|
|
# another object replace the first one
|
|
sheerka.set_in_memory(context, "a", "foo")
|
|
assert sheerka.om.copy(SheerkaMemory.OBJECTS_ENTRY) == {"a": MemoryObject(context.event.get_digest(),
|
|
context.event.date.timestamp(),
|
|
"foo")}
|
|
|
|
def test_i_can_use_memory_with_a_string(self):
|
|
sheerka, context = self.init_test().unpack()
|
|
|
|
foo = Concept("foo")
|
|
sheerka.add_to_memory(context, "foo", foo)
|
|
|
|
assert sheerka.memory(context, "foo") == foo
|
|
|
|
def test_i_can_use_memory_with_a_concept(self):
|
|
sheerka, context = self.init_test().unpack()
|
|
|
|
foo = Concept("foo")
|
|
sheerka.add_to_memory(context, "foo", foo)
|
|
|
|
assert sheerka.memory(context, Concept("foo")) == foo
|
|
|
|
def test_i_can_use_memory_with_a_query(self):
|
|
sheerka, context, foo, bar = self.init_concepts("foo", "bar")
|
|
|
|
sheerka.add_to_memory(context, "x", foo)
|
|
sheerka.add_to_memory(context, "y", bar)
|
|
|
|
assert sheerka.memory(context, "self.name == 'foo'") == foo
|
|
|
|
def test_i_can_use_memory_when_the_entry_does_not_exist(self):
|
|
sheerka, context, foo, bar = self.init_test().with_concepts("foo", "bar", create_new=True).unpack()
|
|
|
|
sheerka.add_to_memory(context, "foo", foo) # add at least one element
|
|
|
|
res = sheerka.memory(context, "bar")
|
|
assert sheerka.isinstance(res, BuiltinConcepts.NOT_FOUND)
|
|
assert res.body == {'#name': 'bar'}
|
|
|
|
def test_i_retrieve_the_last_entry_when_requesting_memory_with_a_query(self):
|
|
sheerka, context, foo, bar, foo2 = self.init_concepts("foo", "bar", Concept("foo", body="2"))
|
|
|
|
sheerka.add_to_memory(context, "x", foo)
|
|
sheerka.add_to_memory(context, "y", bar)
|
|
|
|
context2 = self.get_context(sheerka) # timestamp is newer
|
|
sheerka.add_to_memory(context2, "z", foo2)
|
|
|
|
assert sheerka.memory(context, "self.name == 'foo'") == foo2
|
|
|
|
def test_i_can_look_in_the_previous_objects_when_using_query_to_request_the_memory(self):
|
|
sheerka, context, foo, bar, baz = self.init_concepts("foo", "bar", "baz")
|
|
|
|
sheerka.add_to_memory(context, "x", foo)
|
|
sheerka.add_to_memory(context, "y", bar)
|
|
sheerka.add_to_memory(context, "z", baz)
|
|
|
|
# another layer
|
|
sheerka.add_to_memory(context, "x", bar)
|
|
sheerka.add_to_memory(context, "y", baz)
|
|
|
|
# another layer
|
|
sheerka.add_to_memory(context, "x", baz)
|
|
|
|
# so under x there is [foo] -> [bar] -> [baz]
|
|
# so under y there is [bar] -> [baz]
|
|
# so under z there is [baz]
|
|
assert sheerka.memory(context, "self.name == 'foo'") == foo
|
|
|
|
def test_concept_not_found_is_return_when_not_found(self):
|
|
sheerka, context = self.init_test().unpack()
|
|
|
|
res = sheerka.memory(context, "foo")
|
|
|
|
assert sheerka.isinstance(res, BuiltinConcepts.NOT_FOUND)
|
|
assert res.body == {"#name": "foo"}
|
|
|
|
def test_memory_only_returns_the_last_object(self):
|
|
sheerka, context = self.init_test().unpack()
|
|
|
|
foo = Concept("foo")
|
|
bar = Concept("bar")
|
|
|
|
sheerka.add_to_memory(context, "item", foo)
|
|
sheerka.add_to_memory(context, "item", bar)
|
|
|
|
assert sheerka.memory(context, "item") == bar
|
|
|
|
def test_object_are_not_added_in_memory_during_the_initialisation(self):
|
|
sheerka, context = self.init_test().unpack()
|
|
|
|
assert len(sheerka.om.get_all(SheerkaMemory.OBJECTS_ENTRY)) == 0
|
|
|
|
def test_adding_the_same_object_with_same_context_has_no_effect_when_one_element(self):
|
|
sheerka, context, foo = self.init_concepts("foo")
|
|
service = sheerka.services[SheerkaMemory.NAME]
|
|
|
|
sheerka.add_to_memory(context, "item", foo)
|
|
from_memory = service.get_from_memory(context, "item")
|
|
|
|
sheerka.add_to_memory(context, "item", foo)
|
|
from_memory_again = service.get_from_memory(context, "item")
|
|
assert from_memory_again == from_memory
|
|
|
|
def test_adding_the_same_object_with_same_context_has_no_effect_when_multiple_elements(self):
|
|
sheerka, context, foo, bar, baz = self.init_concepts("foo", "bar", "baz")
|
|
service = sheerka.services[SheerkaMemory.NAME]
|
|
|
|
sheerka.add_to_memory(context, "item", foo)
|
|
sheerka.add_to_memory(context, "item", bar)
|
|
sheerka.add_to_memory(context, "item", baz)
|
|
from_memory = service.get_from_memory(context, "item").copy()
|
|
|
|
sheerka.add_to_memory(context, "item", baz)
|
|
from_memory_again = service.get_from_memory(context, "item")
|
|
assert from_memory_again == from_memory
|
|
|
|
def test_adding_the_same_object_with_a_different_context_updates_event_id_when_one_element(self):
|
|
sheerka, context1, foo = self.init_concepts("foo")
|
|
service = sheerka.services[SheerkaMemory.NAME]
|
|
|
|
sheerka.add_to_memory(context1, "item", foo)
|
|
from_memory = service.get_from_memory(context1, "item")
|
|
|
|
context2 = self.get_context(self.sheerka, message="another message")
|
|
sheerka.add_to_memory(context2, "item", foo)
|
|
from_memory_again = service.get_from_memory(context2, "item")
|
|
|
|
assert from_memory_again != from_memory
|
|
assert not isinstance(from_memory_again, list)
|
|
|
|
assert from_memory_again.obj == from_memory.obj
|
|
assert from_memory_again.event_id == context2.event.get_digest()
|
|
assert from_memory.event_id == context1.event.get_digest()
|
|
|
|
def test_adding_the_same_object_with_a_different_context_updates_event_id_when_multiple_elements(self):
|
|
sheerka, context, foo, bar, baz = self.init_concepts("foo", "bar", "baz")
|
|
service = sheerka.services[SheerkaMemory.NAME]
|
|
|
|
sheerka.add_to_memory(context, "item", foo)
|
|
sheerka.add_to_memory(context, "item", bar)
|
|
sheerka.add_to_memory(context, "item", baz)
|
|
from_memory = service.get_from_memory(context, "item").copy()
|
|
|
|
context2 = self.get_context(self.sheerka, message="another message")
|
|
sheerka.add_to_memory(context2, "item", baz)
|
|
from_memory_again = service.get_from_memory(context2, "item")
|
|
|
|
assert isinstance(from_memory_again, list)
|
|
assert len(from_memory) == len(from_memory_again)
|
|
for mo, mo_again in zip(from_memory, from_memory_again[:-1]):
|
|
assert mo == mo_again
|
|
|
|
assert from_memory[-1].obj == from_memory_again[-1].obj
|
|
assert from_memory[-1].event_id != from_memory_again[-1].event_id
|
|
|
|
def test_object_values_are_recorded(self):
|
|
sheerka, context, foo, prop, value, inner_value = self.init_test(cache_only=False).with_concepts(
|
|
Concept("foo").def_var("x"),
|
|
Concept("prop"),
|
|
Concept("value x").def_var("x"),
|
|
Concept("one", body="1")
|
|
).unpack()
|
|
|
|
instantiated_foo = sheerka.new(foo, x="value1")
|
|
instantiated_one = ensure_evaluated(context, sheerka.new(inner_value))
|
|
instantiated_value = sheerka.new(value, x=instantiated_one)
|
|
sheerka.set_attr(context, instantiated_foo, prop, instantiated_value)
|
|
|
|
sheerka.add_to_memory(context, "foo", instantiated_foo)
|
|
sheerka.om.commit(context)
|
|
|
|
from_db = sheerka.om.current_sdp().get(SheerkaMemory.OBJECTS_ENTRY, "foo")
|
|
assert isinstance(from_db, MemoryObject)
|
|
assert sheerka.isinstance(from_db.obj, foo)
|
|
compare_with_test_object(sheerka.get_attr(from_db.obj, prop), CV(value, x=CB(inner_value, 1)))
|
|
|
|
|
|
class TestSheerkaMemoryUsingFileBase(TestUsingFileBasedSheerka):
|
|
def test_i_can_record_memory_objects(self):
|
|
sheerka, context = self.init_test().unpack()
|
|
|
|
sheerka.add_to_memory(context, "item", Concept("foo"))
|
|
sheerka.om.commit(context)
|
|
|
|
sheerka = self.get_sheerka()
|
|
context = self.get_context(sheerka)
|
|
assert sheerka.get_from_memory(context, "item").obj == Concept("foo")
|