from core.builtin_concepts import BuiltinConcepts from core.concept import Concept from core.sheerka.ExecutionContext import ExecutionContext from core.sheerka.services.SheerkaMemory import SheerkaMemory from sdp.sheerkaDataProvider import Event from tests.TestUsingMemoryBasedSheerka import TestUsingMemoryBasedSheerka class TestExecutionContext(TestUsingMemoryBasedSheerka): def test_id_is_incremented_by_event_digest(self): sheerka = self.get_sheerka() a = ExecutionContext("foo", Event("event_1"), sheerka, BuiltinConcepts.NOP, None) b = ExecutionContext("foo", Event("event_1"), sheerka, BuiltinConcepts.NOP, None) c = ExecutionContext("foo", Event("event_2"), sheerka, BuiltinConcepts.NOP, None) d = b.push(BuiltinConcepts.NOP, None) e = c.push(BuiltinConcepts.NOP, None) assert a.id == 0 assert b.id == 1 assert c.id == 0 assert d.id == 2 assert e.id == 1 def test_i_can_use_with_statement(self): sheerka = self.get_sheerka() with ExecutionContext("who_", Event("event"), sheerka, BuiltinConcepts.NOP, None) as e: pass assert e.elapsed > 0 def test_i_can_push(self): sheerka = self.get_sheerka() a = ExecutionContext("foo", Event("event_1"), sheerka, BuiltinConcepts.NOP, None, desc="some description", obj=Concept("foo"), concepts={"bar": Concept("bar")}) a.preprocess = set() a.preprocess.add("preprocess") a.preprocess_parsers = ["list of parsers"] a.preprocess_evaluators = ["list of evaluators"] a.protected_hints.add(BuiltinConcepts.EVAL_BODY_REQUESTED) a.global_hints.add(BuiltinConcepts.EVAL_BODY_REQUESTED) b = a.push(BuiltinConcepts.EVALUATION, "sub action context", desc="sub description") assert b._parent == a assert b.who == a.who assert b.event == a.event assert b.sheerka == a.sheerka assert b.action == BuiltinConcepts.EVALUATION assert b.action_context == "sub action context" assert b.desc == "sub description" assert b.obj == a.obj assert b.concepts == a.concepts assert b.id == a.id + 1 assert b.preprocess == a.preprocess assert b.preprocess_parsers == a.preprocess_parsers assert b.preprocess_evaluators == a.preprocess_evaluators assert b.protected_hints == {BuiltinConcepts.EVAL_BODY_REQUESTED} assert b.global_hints == a.global_hints def test_children_i_created_when_i_push(self): sheerka = self.get_sheerka() e = ExecutionContext("who_", Event("event"), sheerka, BuiltinConcepts.NOP, None) e.push(BuiltinConcepts.NOP, None, who="a", desc="I do something") e.push(BuiltinConcepts.NOP, None, who="b", desc="oups! I did a again") e.push(BuiltinConcepts.NOP, None, who="c", desc="I do something else") assert len(e._children) == 3 assert e._children[0].who, e._children[0].who == ("a", "I do something") assert e._children[1].who, e._children[1].who == ("b", "oups! I did a again") assert e._children[2].who, e._children[2].who == ("c", "I do something else") # def test_i_can_add_variable_when_i_push(self): # sheerka = self.get_sheerka() # # e = ExecutionContext("who_", Event("event"), sheerka, BuiltinConcepts.NOP, None) # sub_e = e.push(BuiltinConcepts.NOP, None, who="a", my_new_var="new var value") # # assert sub_e.my_new_var == "new var value" # with pytest.raises(AttributeError): # assert e.my_new_var == "" # my_new_var does not exist in parent def test_local_hints_are_local_and_global_hints_are_global(self): sheerka = self.get_sheerka() a = ExecutionContext("foo", Event("event_1"), sheerka, BuiltinConcepts.NOP, None) a.protected_hints.add("local hint 1") a.global_hints.add("global hint 1") b = a.push(BuiltinConcepts.NOP, None) b.protected_hints.add("local hint 2") b.global_hints.add("global hint 2") assert a.protected_hints == {"local hint 1"} assert a.global_hints == {"global hint 1", "global hint 2"} assert b.protected_hints == {"local hint 1", "local hint 2"} assert b.global_hints == {"global hint 1", "global hint 2"} def test_global_hits_are_global_even_when_empty(self): sheerka = self.get_sheerka() a = ExecutionContext("foo", Event("event_1"), sheerka, BuiltinConcepts.NOP, None) b = a.push(BuiltinConcepts.NOP, None) b.global_hints.add("global hint 2") assert a.global_hints == {"global hint 2"} assert b.global_hints == {"global hint 2"} def test_i_can_search(self): sheerka = self.get_sheerka() a = ExecutionContext("foo", Event("event_1"), sheerka, BuiltinConcepts.TESTING, "a") ab = a.push(BuiltinConcepts.TESTING, "ab", obj="obj_ab") ac = a.push(BuiltinConcepts.TESTING, "ac", obj="obj_ac") abb = ab.push(BuiltinConcepts.TESTING, "abb", obj="skip") abbb = abb.push(BuiltinConcepts.TESTING, "abbb", obj="obj_abbb") assert list(abbb.search()) == [abb, ab, a] assert list(abbb.search(start_with_self=True)) == [abbb, abb, ab, a] assert list(abbb.search(lambda ec: ec.obj != "skip")) == [ab, a] assert list(abbb.search(lambda ec: ec.obj != "skip", lambda ec: ec.action_context)) == ["ab", "a"] assert list(abbb.search(stop=lambda ec: ec.obj == "skip")) == [abb] assert list(abbb.search( stop=lambda ec: ec.obj == "skip", start_with_self=True, get_obj=lambda ec: ec.obj)) == ["obj_abbb", "skip"] def test_i_can_deactivate_push(self): sheerka = self.get_sheerka() context = ExecutionContext("foo", Event("event_1"), sheerka, BuiltinConcepts.NOP, None) context.deactivate_push() sub_context1 = context.push(BuiltinConcepts.NOP, None) sub_context2 = context.push(BuiltinConcepts.NOP, None) sub_context3 = sub_context1.push(BuiltinConcepts.NOP, None) assert id(sub_context1) == id(sub_context2) assert id(sub_context1) == id(sub_context3) def test_stm_are_copied_when_deactivate_push(self): sheerka = self.get_sheerka() context = ExecutionContext("foo", Event("event_1"), sheerka, BuiltinConcepts.NOP, None) context.add_to_short_term_memory("key1", "value1") context.add_to_short_term_memory("key2", "value2") context.deactivate_push() stm_cache = sheerka.services[SheerkaMemory.NAME].short_term_objects.cache assert stm_cache == { context.id: {'key1': 'value1', 'key2': 'value2'}, context.id + 1: {'key1': 'value1', 'key2': 'value2'} } def test_has_parent(self): sheerka = self.get_sheerka() root = ExecutionContext("foo", Event("event_1"), sheerka, BuiltinConcepts.NOP, None) sub1 = root.push(BuiltinConcepts.TESTING, None) sub2 = root.push(BuiltinConcepts.TESTING, None) sub11 = sub1.push(BuiltinConcepts.TESTING, None) sub111 = sub11.push(BuiltinConcepts.TESTING, None) assert not root.has_parent(sub1.id) assert sub1.has_parent(root.id) assert sub11.has_parent(root.id) assert sub111.has_parent(root.id) assert sub2.has_parent(root.id) assert not sub1.has_parent(sub2.id) assert not sub2.has_parent(sub1.id) # def test_variables_are_passed_to_children_but_not_to_parents(self): # sheerka = self.get_sheerka() # # a = ExecutionContext("foo", Event("event_1"), sheerka, BuiltinConcepts.NOP, None) # assert not hasattr(a, "var") # # b = a.push(BuiltinConcepts.NOP, None, var="foo") # assert b.var == "foo" # assert not hasattr(a, "var") # # c = b.push(BuiltinConcepts.NOP, None) # assert c.var == "foo" # # c.var = "bar" # assert c.var == "bar" # assert b.var != "bar" # assert not hasattr(a, "var")