kodjo
945807b375
Fixed #74 : Keyword parameters are no longer recognized when a concept that redefines equality is created Fixed #118 : RecursionError: maximum recursion depth exceeded Fixed #119 : PreventCircularReferenceEvaluator Fixed #121 : Plural are not updated when new elements are added Fixed #123 : BaseCache : Values in cache can be evicted before being committed Fixed #105 : TOO_MANY_ERROR is not the relevant error when results are filtered
236 lines
9.9 KiB
Python
236 lines
9.9 KiB
Python
from core.builtin_concepts_ids 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()
|
|
|
|
event1 = Event("event_1")
|
|
event2 = Event("event_2")
|
|
|
|
a = ExecutionContext("foo", event1, sheerka, BuiltinConcepts.NOP, None)
|
|
b = ExecutionContext("foo", event1, sheerka, BuiltinConcepts.NOP, None)
|
|
c = ExecutionContext("foo", event2, 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)
|
|
a.possible_variables = {"a", "b", "c"}
|
|
|
|
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
|
|
assert b.possible_variables == a.possible_variables
|
|
|
|
def test_children_are_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].desc) == ("a", "I do something")
|
|
assert (e._children[1].who, e._children[1].desc) == ("b", "oups! I did a again")
|
|
assert (e._children[2].who, e._children[2].desc) == ("c", "I do something else")
|
|
|
|
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"]
|
|
assert list(abbb.search(only_first=True)) == [abb]
|
|
assert list(abbb.search(start_with_self=True, only_first=True)) == [abbb]
|
|
assert list(abbb.search(lambda ec: ec.obj != "skip", only_first=True)) == [ab]
|
|
|
|
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_i_can_reset_global_hints(self):
|
|
sheerka = self.get_sheerka()
|
|
|
|
context = ExecutionContext("foo", Event("event_1"), sheerka, BuiltinConcepts.NOP, None)
|
|
context.add_to_global_hints(BuiltinConcepts.TESTING)
|
|
context.add_to_global_hints(BuiltinConcepts.DEBUG)
|
|
|
|
sub_context1 = context.push(BuiltinConcepts.NOP, None)
|
|
assert BuiltinConcepts.TESTING in sub_context1.global_hints
|
|
assert BuiltinConcepts.DEBUG in sub_context1.global_hints
|
|
|
|
sub_context2 = context.push(BuiltinConcepts.NOP, None, reset_hints={BuiltinConcepts.TESTING})
|
|
assert BuiltinConcepts.TESTING not in sub_context2.global_hints
|
|
assert BuiltinConcepts.DEBUG in sub_context2.global_hints
|
|
|
|
sub_context3 = context.push(BuiltinConcepts.NOP,
|
|
None,
|
|
reset_hints={BuiltinConcepts.DEBUG, BuiltinConcepts.TESTING})
|
|
assert sub_context3.global_hints == set()
|
|
|
|
def test_i_can_reset_protected_hints(self):
|
|
sheerka = self.get_sheerka()
|
|
|
|
context = ExecutionContext("foo", Event("event_1"), sheerka, BuiltinConcepts.NOP, None)
|
|
context.add_to_protected_hints(BuiltinConcepts.TESTING)
|
|
context.add_to_protected_hints(BuiltinConcepts.DEBUG)
|
|
|
|
sub_context1 = context.push(BuiltinConcepts.NOP, None)
|
|
assert BuiltinConcepts.TESTING in sub_context1.protected_hints
|
|
assert BuiltinConcepts.DEBUG in sub_context1.protected_hints
|
|
|
|
sub_context2 = context.push(BuiltinConcepts.NOP, None, reset_hints={BuiltinConcepts.TESTING})
|
|
assert BuiltinConcepts.TESTING not in sub_context2.protected_hints
|
|
assert BuiltinConcepts.DEBUG in sub_context2.protected_hints
|
|
|
|
sub_context3 = context.push(BuiltinConcepts.NOP,
|
|
None,
|
|
reset_hints={BuiltinConcepts.DEBUG, BuiltinConcepts.TESTING})
|
|
assert sub_context3.protected_hints == set()
|
|
|
|
# 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")
|