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 SheerkaMemory, MemoryObject from tests.TestUsingFileBasedSheerka import TestUsingFileBasedSheerka from tests.TestUsingMemoryBasedSheerka import TestUsingMemoryBasedSheerka from tests.parsers.parsers_utils import CV, compare_with_test_object, CB 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(), foo)} assert id(sheerka.get_from_memory(context, "a").obj) == id(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_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(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")