318 lines
12 KiB
Python
318 lines
12 KiB
Python
import os
|
|
|
|
import pytest
|
|
from core.builtin_concepts import ParserResultConcept, ReturnValueConcept, BuiltinConcepts
|
|
from core.concept import Concept
|
|
from core.sheerka.ExecutionContext import ExecutionContext
|
|
from evaluators.ExplainEvaluator import ExplainEvaluator
|
|
from parsers.ExplainParser import ExplanationNode, RecurseDefNode, FormatLNode, UnionNode, FilterNode, FormatDNode
|
|
from parsers.ExpressionParser import PropertyEqualsNode, PropertyEqualsSequenceNode, TrueNode, IsaNode
|
|
from printer.FormatInstructions import FormatDetailDesc, FormatDetailType
|
|
from pytest import fixture
|
|
from sdp.sheerkaDataProvider import Event
|
|
from sdp.sheerkaSerializer import Serializer, SerializerContext
|
|
|
|
from tests.TestUsingMemoryBasedSheerka import TestUsingMemoryBasedSheerka
|
|
|
|
|
|
@fixture(scope="module")
|
|
def serializer():
|
|
"""
|
|
Return a :class:`sdp.sheerkaSerializer.Serializer` instance for the module
|
|
"""
|
|
return Serializer()
|
|
|
|
|
|
class EC:
|
|
"""
|
|
Helper to create execution context (AKA execution result)
|
|
"""
|
|
|
|
def __init__(self, children=None, **props):
|
|
self.props = props
|
|
self.children = children
|
|
|
|
|
|
def get_return_value(expr):
|
|
if isinstance(expr, ExplanationNode):
|
|
value = expr
|
|
else:
|
|
value = ExplanationNode("xxx_test_explain_evaluator_xxx", "", expr=expr)
|
|
|
|
return ReturnValueConcept(
|
|
"TestEvaluator",
|
|
True,
|
|
ParserResultConcept(parser="parser", value=value))
|
|
|
|
|
|
def create_executions_results(context, list_of_ecs):
|
|
def update(execution_context, ec):
|
|
for prop_name, pro_value in ec.props.items():
|
|
setattr(execution_context, prop_name, pro_value)
|
|
|
|
if ec.children:
|
|
for child_ec in ec.children:
|
|
child_execution_context = execution_context.push("TestEvaluator")
|
|
update(child_execution_context, child_ec)
|
|
|
|
res = []
|
|
for ec in list_of_ecs:
|
|
execution_context = ExecutionContext("TestEvaluator", context.event, context.sheerka)
|
|
update(execution_context, ec)
|
|
res.append(execution_context)
|
|
|
|
return res
|
|
|
|
|
|
def get_execution_result_from_file(sheerka, digest, serializer):
|
|
target_path = os.path.join("../_fixture/", digest) + "_result"
|
|
with open(target_path, "rb") as f:
|
|
context = SerializerContext(sheerka=sheerka)
|
|
return serializer.deserialize(f, context)
|
|
|
|
|
|
def get_execution_result_from_list(executions_result):
|
|
return executions_result
|
|
|
|
|
|
class TestExplainEvaluator(TestUsingMemoryBasedSheerka):
|
|
|
|
@staticmethod
|
|
def init_evaluator_with_file(self, serializer):
|
|
sheerka = self.get_sheerka()
|
|
context = self.get_context(sheerka)
|
|
evaluator = ExplainEvaluator()
|
|
evaluator.get_execution_result = lambda s, d: get_execution_result_from_file(s, d, serializer)
|
|
|
|
return sheerka, context, evaluator
|
|
|
|
def init_evaluator_with_list(self, list_of_ecs):
|
|
sheerka = self.get_sheerka()
|
|
context = self.get_context(sheerka)
|
|
evaluator = ExplainEvaluator()
|
|
|
|
executions_result = create_executions_results(context, list_of_ecs)
|
|
evaluator.get_execution_result = lambda s, d: get_execution_result_from_list(executions_result)
|
|
|
|
return sheerka, context, evaluator, executions_result
|
|
|
|
@pytest.mark.parametrize("ret_val, expected", [
|
|
(ReturnValueConcept("some_name", True, ParserResultConcept(value=ExplanationNode("", ""))), True),
|
|
(ReturnValueConcept("some_name", True, ParserResultConcept(value="other thing")), False),
|
|
(ReturnValueConcept("some_name", False, "not relevant"), False),
|
|
(ReturnValueConcept("some_name", True, Concept()), False)
|
|
])
|
|
def test_i_can_match(self, ret_val, expected):
|
|
context = self.get_context()
|
|
assert ExplainEvaluator().matches(context, ret_val) == expected
|
|
|
|
def test_i_can_eval_in_list(self, serializer):
|
|
sheerka, context, evaluator, execution_results = self.init_evaluator_with_list(
|
|
[
|
|
EC(desc="correct desc"),
|
|
EC(desc="wrong desc"),
|
|
]
|
|
)
|
|
ret_val = get_return_value(UnionNode(
|
|
[
|
|
FilterNode(TrueNode()),
|
|
FilterNode(PropertyEqualsNode("desc", "correct desc")),
|
|
]))
|
|
|
|
res = evaluator.eval(context, ret_val)
|
|
assert res.status
|
|
assert sheerka.isinstance(res.body, BuiltinConcepts.EXPLANATION)
|
|
|
|
filtered = res.body.body
|
|
assert filtered == [execution_results[0]]
|
|
|
|
def test_i_can_eval_in_children(self):
|
|
sheerka, context, evaluator, execution_results = self.init_evaluator_with_list(
|
|
[
|
|
EC(desc="wrong desc", children=[EC(desc="wrong sub"), EC(desc="good sub")]),
|
|
EC(desc="wrong desc", children=[EC(desc="good sub")]),
|
|
]
|
|
)
|
|
ret_val = get_return_value(UnionNode(
|
|
[
|
|
FilterNode(TrueNode()),
|
|
FilterNode(PropertyEqualsNode("desc", "good sub")),
|
|
]))
|
|
|
|
res = evaluator.eval(context, ret_val)
|
|
assert res.status
|
|
assert sheerka.isinstance(res.body, BuiltinConcepts.EXPLANATION)
|
|
|
|
filtered = res.body.body
|
|
assert filtered == [
|
|
execution_results[0].children[1],
|
|
execution_results[1].children[0],
|
|
]
|
|
|
|
def test_i_can_evaluate_multiple_filter_node(self):
|
|
sheerka, context, evaluator, execution_results = self.init_evaluator_with_list(
|
|
[
|
|
EC(desc="parent1", _id=1, children=[EC(desc="wrong sub"), EC(desc="good sub")]),
|
|
EC(desc="parent2", children=[EC(desc="wrong sub"), EC(desc="good sub")]),
|
|
EC(desc="good sub")
|
|
])
|
|
ret_val = get_return_value(UnionNode(
|
|
[
|
|
FilterNode(TrueNode()),
|
|
FilterNode(PropertyEqualsNode("id", "1")),
|
|
FilterNode(PropertyEqualsNode("desc", "good sub")),
|
|
]))
|
|
|
|
res = evaluator.eval(context, ret_val)
|
|
assert res.status
|
|
assert len(res.body) == 2
|
|
|
|
assert sheerka.isinstance(res.body[0], BuiltinConcepts.EXPLANATION)
|
|
assert sheerka.isinstance(res.body[1], BuiltinConcepts.EXPLANATION)
|
|
|
|
assert res.body[0].body == [execution_results[0]]
|
|
assert res.body[1].body == [
|
|
execution_results[0].children[1],
|
|
execution_results[1].children[1],
|
|
execution_results[2]
|
|
]
|
|
|
|
def test_i_can_eval_parent_and_child(self):
|
|
sheerka, context, evaluator, execution_results = self.init_evaluator_with_list(
|
|
[
|
|
EC(desc="parent1", children=[EC(desc="wrong sub"), EC(desc="good sub")]),
|
|
EC(desc="parent2", children=[EC(desc="wrong sub"), EC(desc="good sub")]),
|
|
EC(desc="good sub")
|
|
]
|
|
)
|
|
ret_val = get_return_value(UnionNode(
|
|
[
|
|
FilterNode(TrueNode()),
|
|
FilterNode(PropertyEqualsSequenceNode(["desc", "desc"], ["parent1", "good sub"])),
|
|
]))
|
|
|
|
res = evaluator.eval(context, ret_val)
|
|
assert res.status
|
|
assert sheerka.isinstance(res.body, BuiltinConcepts.EXPLANATION)
|
|
|
|
filtered = res.body.body
|
|
assert filtered == [
|
|
execution_results[0].children[1],
|
|
]
|
|
|
|
def test_i_correctly_create_format_instructions(self):
|
|
sheerka, context, evaluator, execution_results = self.init_evaluator_with_list([])
|
|
ret_val = get_return_value(UnionNode(
|
|
[
|
|
FilterNode(TrueNode(), [
|
|
RecurseDefNode(2),
|
|
FormatLNode("abc"),
|
|
FormatDNode({"a": "{a}", "b": "{b}"})
|
|
]),
|
|
]))
|
|
|
|
res = evaluator.eval(context, ret_val)
|
|
assert res.status
|
|
assert sheerka.isinstance(res.body, BuiltinConcepts.EXPLANATION)
|
|
|
|
instructions = res.body.instructions
|
|
assert instructions.recursive_props == {"children": 2}
|
|
assert instructions.format_l == {'core.sheerka.ExecutionContext.ExecutionContext': 'abc'}
|
|
assert instructions.format_d == [FormatDetailDesc(
|
|
IsaNode(ExecutionContext),
|
|
FormatDetailType.Props_In_Line,
|
|
{"a": "{a}", "b": "{b}"})]
|
|
|
|
def test_i_correctly_create_format_instructions_with_filtering(self):
|
|
sheerka, context, evaluator, execution_results = self.init_evaluator_with_list([])
|
|
|
|
ret_val = get_return_value(UnionNode(
|
|
[
|
|
FilterNode(TrueNode()),
|
|
FilterNode(PropertyEqualsNode("id", "1"), [RecurseDefNode(2), FormatLNode("abc")]),
|
|
]))
|
|
res = evaluator.eval(context, ret_val)
|
|
assert res.status
|
|
assert sheerka.isinstance(res.body, BuiltinConcepts.EXPLANATION)
|
|
|
|
instructions = res.body.instructions
|
|
assert instructions.format_l == {'core.sheerka.ExecutionContext.ExecutionContext': 'abc'}
|
|
assert instructions.recursive_props == {"children": 2}
|
|
|
|
def test_i_can_have_different_instructions_for_different_filtering(self):
|
|
sheerka, context, evaluator, execution_results = self.init_evaluator_with_list([])
|
|
ret_val = get_return_value(UnionNode(
|
|
[
|
|
FilterNode(TrueNode()),
|
|
FilterNode(PropertyEqualsNode("id", "1"), [RecurseDefNode(2)]),
|
|
FilterNode(PropertyEqualsNode("desc", "good sub"), [FormatLNode("abc")]),
|
|
]))
|
|
|
|
res = evaluator.eval(context, ret_val)
|
|
assert res.status
|
|
assert len(res.body) == 2
|
|
|
|
assert res.body[0].instructions.recursive_props == {"children": 2}
|
|
assert res.body[1].instructions.format_l == {'core.sheerka.ExecutionContext.ExecutionContext': 'abc'}
|
|
|
|
def test_filtering_instructions_inherit_from_the_first_filtering_node(self):
|
|
sheerka, context, evaluator, execution_results = self.init_evaluator_with_list([])
|
|
ret_val = get_return_value(UnionNode(
|
|
[
|
|
FilterNode(TrueNode(), [RecurseDefNode(2)]),
|
|
FilterNode(PropertyEqualsNode("id", "1"), [RecurseDefNode(1)]),
|
|
FilterNode(PropertyEqualsNode("desc", "good sub"), [FormatLNode("abc")]),
|
|
]))
|
|
|
|
res = evaluator.eval(context, ret_val)
|
|
assert res.status
|
|
assert len(res.body) == 2
|
|
|
|
assert res.body[0].instructions.recursive_props == {"children": 1} # overridden
|
|
|
|
assert res.body[1].instructions.format_l == {'core.sheerka.ExecutionContext.ExecutionContext': 'abc'}
|
|
assert res.body[1].instructions.recursive_props == {"children": 2}
|
|
|
|
def test_i_can_reuse_a_recorded_digest(self):
|
|
sheerka, context, evaluator, execution_results = self.init_evaluator_with_list([])
|
|
expr = UnionNode([FilterNode(TrueNode(), [RecurseDefNode(2)])])
|
|
|
|
# need a valid result to test this feature
|
|
event = Event("fake message")
|
|
execution_context = ExecutionContext("TestExplainEvaluator", event, sheerka)
|
|
sheerka.sdp.save_result(execution_context)
|
|
|
|
# save another result
|
|
event2 = Event("fake message")
|
|
execution_context = ExecutionContext("TestExplainEvaluator", event2, sheerka)
|
|
sheerka.sdp.save_result(execution_context)
|
|
|
|
# digest is recorded during the first call
|
|
explanation_node = ExplanationNode(event.get_digest(), "", expr=expr, record_digest=True)
|
|
ret_val = get_return_value(explanation_node)
|
|
evaluator.eval(context, ret_val)
|
|
|
|
# the next call to get_event_digest will load the recorded digest
|
|
explanation_node = ExplanationNode("", "", expr=expr, record_digest=False) # digest is not provided
|
|
digest = evaluator.get_event_digest(sheerka, explanation_node)
|
|
assert digest == event.get_digest()
|
|
|
|
# test I can record another digest
|
|
explanation_node = ExplanationNode(event2.get_digest(), "", expr=expr, record_digest=True)
|
|
ret_val = get_return_value(explanation_node)
|
|
evaluator.eval(context, ret_val)
|
|
|
|
explanation_node = ExplanationNode("", "", expr=expr, record_digest=False) # digest is not provided
|
|
digest = evaluator.get_event_digest(sheerka, explanation_node)
|
|
assert digest == event2.get_digest()
|
|
|
|
# test can now reset the recorded digest
|
|
# (a digest is provided, but record_digest is set to False)
|
|
explanation_node = ExplanationNode(event.get_digest(), "", expr=expr, record_digest=False)
|
|
ret_val = get_return_value(explanation_node)
|
|
evaluator.eval(context, ret_val)
|
|
|
|
explanation_node = ExplanationNode("", "", expr=expr, record_digest=False) # digest is not provided
|
|
digest = evaluator.get_event_digest(sheerka, explanation_node)
|
|
assert digest is None
|