382 lines
20 KiB
Python
382 lines
20 KiB
Python
import pytest
|
|
from core.builtin_concepts import ReturnValueConcept, BuiltinConcepts
|
|
from core.concept import Concept
|
|
from core.rule import Rule
|
|
from core.sheerka.services.SheerkaOut import SheerkaOut
|
|
from core.sheerka.services.SheerkaRuleManager import FormatAstRawText, FormatAstVariable, FormatAstSequence, \
|
|
FormatAstColor, FormatAstVariableNotFound, FormatAstList
|
|
from core.utils import flatten_all_children
|
|
|
|
from tests.TestUsingMemoryBasedSheerka import TestUsingMemoryBasedSheerka
|
|
|
|
seq = FormatAstSequence
|
|
raw = FormatAstRawText
|
|
var = FormatAstVariable
|
|
|
|
|
|
class TestSheerkaOut(TestUsingMemoryBasedSheerka):
|
|
|
|
def init_service_with_rules(self, *rules, **kwargs):
|
|
sheerka, context, *rules = self.init_format_rules(*rules, **kwargs)
|
|
service = sheerka.services[SheerkaOut.NAME]
|
|
|
|
return sheerka, context, service, *rules
|
|
|
|
@pytest.mark.parametrize("rule, expected", [
|
|
(("__ret", "hello world"), FormatAstRawText("hello world")),
|
|
(("__ret", "{__ret_value}"), FormatAstVariable("__ret_value", None, 1)),
|
|
(("__ret", "{status}"), FormatAstVariable("status", None, True)),
|
|
(("__ret", "{foo}"), FormatAstVariableNotFound("foo")),
|
|
(("__ret", "hello world {__ret_value} !"), seq([FormatAstRawText("hello world "),
|
|
FormatAstVariable("__ret_value", None, 1),
|
|
FormatAstRawText(" !")])),
|
|
(("__ret", "red(__ret_value)"), FormatAstColor("red", FormatAstVariable("__ret_value", None, 1))),
|
|
(("__ret", "blue('hello world {__ret_value} !')"), FormatAstColor("blue",
|
|
seq([FormatAstRawText("hello world "),
|
|
FormatAstVariable("__ret_value", None,
|
|
1),
|
|
FormatAstRawText(" !")]))),
|
|
(("__ret", "list(foo)"), FormatAstVariableNotFound("foo")),
|
|
(("__ret", "{__ret_value:3}"), FormatAstVariable("__ret_value", "3", " 1"))
|
|
])
|
|
def test_i_can_develop_when_simple_rules(self, rule, expected):
|
|
sheerka, context, service, rule = self.init_service_with_rules(rule)
|
|
ret = sheerka.ret("Test", True, 1)
|
|
|
|
res = service.create_out_tree(context, ret)
|
|
|
|
assert res == expected
|
|
|
|
def test_i_can_develop_rules_of_rules_using_obj_type(self):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(
|
|
("__ret", "status: {status}, value: {__ret_value}"), # ret_value is a 'foo'
|
|
("isinstance(__obj, 'foo')", "{id}-{name}:{body}")
|
|
)
|
|
sheerka.create_new_concept(context, Concept("foo"))
|
|
foo = sheerka.new("foo", body="hello world")
|
|
ret = sheerka.ret("Test", True, foo)
|
|
|
|
res = service.create_out_tree(context, ret)
|
|
|
|
assert res == seq([FormatAstRawText("status: "),
|
|
FormatAstVariable("status", None, True),
|
|
FormatAstRawText(", value: "),
|
|
FormatAstVariable("__ret_value", None, seq([FormatAstVariable("id", None, "1001"),
|
|
FormatAstRawText("-"),
|
|
FormatAstVariable("name", None, "foo"),
|
|
FormatAstRawText(":"),
|
|
FormatAstVariable("body", None,
|
|
"hello world")]))])
|
|
|
|
@pytest.mark.parametrize('second_rule', [
|
|
("__ret.body", "{id}-{name}:{body}"),
|
|
("body", "{id}-{name}:{body}"),
|
|
])
|
|
def test_i_can_develop_rules_of_rules_using_obj_names(self, second_rule):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(
|
|
("__ret", "status: {status}, value: {__ret.body}"),
|
|
second_rule
|
|
)
|
|
sheerka.create_new_concept(context, Concept("foo"))
|
|
foo = sheerka.new("foo", body="hello world")
|
|
ret = sheerka.ret("Test", True, foo)
|
|
|
|
res = service.create_out_tree(context, ret)
|
|
|
|
assert res == seq([FormatAstRawText("status: "),
|
|
FormatAstVariable("status", None, True),
|
|
FormatAstRawText(", value: "),
|
|
FormatAstVariable("__ret.body", None, seq([FormatAstVariable("id", None, "1001"),
|
|
FormatAstRawText("-"),
|
|
FormatAstVariable("name", None, "foo"),
|
|
FormatAstRawText(":"),
|
|
FormatAstVariable("body", None,
|
|
"hello world")]))])
|
|
|
|
def test_i_can_develop_list(self):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(("isinstance(__obj, list)", "list(__obj)"))
|
|
lst = list(flatten_all_children(context, lambda item: item.achildren))[:3]
|
|
|
|
res = service.create_out_tree(context, lst)
|
|
assert res == FormatAstList(variable="__obj", items=[FormatAstVariable("__item", None, lst[0], 0),
|
|
FormatAstVariable("__item", None, lst[1], 1),
|
|
FormatAstVariable("__item", None, lst[2], 2)])
|
|
|
|
def test_i_can_develop_list_using_the_default_items_prop(self):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(("isinstance(__obj, 'foo')", "list(__obj)"))
|
|
lst = list(flatten_all_children(context, lambda item: item.achildren))[:3]
|
|
foo = Concept("foo", body=lst, key="foo").auto_init()
|
|
|
|
res = service.create_out_tree(context, foo)
|
|
assert res == FormatAstList(variable="__obj", items=[FormatAstVariable("__item", None, lst[0], 0),
|
|
FormatAstVariable("__item", None, lst[1], 1),
|
|
FormatAstVariable("__item", None, lst[2], 2)])
|
|
|
|
def test_i_can_develop_list_using_the_custom_items_prop(self):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(
|
|
("isinstance(__obj, 'foo')", "list(__obj, items_prop='custom')"))
|
|
lst = list(flatten_all_children(context, lambda item: item.achildren))[:3]
|
|
foo = Concept("foo", key="foo").def_var("custom").auto_init()
|
|
foo.set_value("custom", lst)
|
|
|
|
res = service.create_out_tree(context, foo)
|
|
assert res == FormatAstList(variable="__obj", items_prop='custom',
|
|
items=[FormatAstVariable("__item", None, lst[0], 0),
|
|
FormatAstVariable("__item", None, lst[1], 1),
|
|
FormatAstVariable("__item", None, lst[2], 2)])
|
|
|
|
def test_not_a_list_are_resolved_into_variable_ast(self):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(("isinstance(__obj, 'foo')", "list(__obj)"))
|
|
foo = Concept("foo", key="foo")
|
|
|
|
res = service.create_out_tree(context, foo)
|
|
assert res == FormatAstVariable("__obj", None, foo)
|
|
|
|
def test_i_can_develop_list_of_return_values(self):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(("__rets", "list(__rets)"))
|
|
lst = [ReturnValueConcept("foo", True, "value for True"),
|
|
ReturnValueConcept("bar", True, Concept("bar", body="value for bar").auto_init()),
|
|
ReturnValueConcept("baz", False, sheerka.new(BuiltinConcepts.ERROR, body="baz in error")),
|
|
]
|
|
|
|
res = service.create_out_tree(context, lst)
|
|
assert res == FormatAstList(variable="__rets", items=[FormatAstVariable("__item", None, lst[0], 0),
|
|
FormatAstVariable("__item", None, lst[1], 1),
|
|
FormatAstVariable("__item", None, lst[2], 2)])
|
|
|
|
def test_rules_are_correctly_reset_when_list(self):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(
|
|
("__rets", "list(__rets)"),
|
|
("__ret and not __ret.status", "red(__ret)")
|
|
)
|
|
lst = [
|
|
ReturnValueConcept("who", False, "value1"),
|
|
ReturnValueConcept("who", False, "value2"),
|
|
]
|
|
|
|
res = service.create_out_tree(context, lst)
|
|
assert res == FormatAstList(variable="__rets", items=[
|
|
FormatAstVariable("__item", None, FormatAstColor("red", FormatAstVariable("__ret", None, lst[0])), 0),
|
|
FormatAstVariable("__item", None, FormatAstColor("red", FormatAstVariable("__ret", None, lst[1])), 1),
|
|
])
|
|
|
|
def test_i_can_develop_list_with_recurse(self):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(
|
|
("__rets", "list(__rets, recurse_on='parents', recursion_depth=2)"))
|
|
r1111 = ReturnValueConcept("Test", True, "r1111")
|
|
r111 = ReturnValueConcept("Test", True, "r111", parents=[r1111])
|
|
r11 = ReturnValueConcept("Test", True, "r11", parents=[r111])
|
|
r1 = ReturnValueConcept("Test", True, "r1", parents=[r11])
|
|
r22 = ReturnValueConcept("Test", True, "r22")
|
|
r2 = ReturnValueConcept("Test", True, "r2", parents=[r22])
|
|
|
|
lst = [r1, r2]
|
|
|
|
res = service.create_out_tree(context, lst)
|
|
assert res == FormatAstList(variable="__rets", recurse_on='parents', recursion_depth=2, items=[
|
|
FormatAstVariable("__item", None, r1, 0),
|
|
FormatAstList(variable="__parents", recurse_on='parents', recursion_depth=1, items=[
|
|
FormatAstVariable("__item", None, r11, 0),
|
|
FormatAstList(variable="__parents", recurse_on='parents', recursion_depth=0, items=[
|
|
FormatAstVariable("__item", None, r111, 0)])
|
|
]),
|
|
FormatAstVariable("__item", None, r2, 1),
|
|
FormatAstList(variable="__parents", recurse_on='parents', recursion_depth=1, items=[
|
|
FormatAstVariable("__item", None, r22, 0)]),
|
|
])
|
|
|
|
def test_i_can_develop_list_with_recurse_using_container_format_instr(self):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(
|
|
("isinstance(__obj, 'foo')", "list(__obj)"))
|
|
r1111 = ReturnValueConcept("Test", True, "r1111")
|
|
r111 = ReturnValueConcept("Test", True, "r111", parents=[r1111])
|
|
r11 = ReturnValueConcept("Test", True, "r11", parents=[r111])
|
|
r1 = ReturnValueConcept("Test", True, "r1", parents=[r11])
|
|
r22 = ReturnValueConcept("Test", True, "r22")
|
|
r2 = ReturnValueConcept("Test", True, "r2", parents=[r22])
|
|
|
|
foo = Concept("foo", key="foo", body=[r1, r2]).auto_init()
|
|
foo.set_format_instr(recurse_on='parents', recursion_depth=2)
|
|
|
|
res = service.create_out_tree(context, foo)
|
|
assert res == FormatAstList(variable="__obj", recurse_on='parents', recursion_depth=2, items=[
|
|
FormatAstVariable("__item", None, r1, 0),
|
|
FormatAstList(variable="__parents", recurse_on='parents', recursion_depth=1, items=[
|
|
FormatAstVariable("__item", None, r11, 0),
|
|
FormatAstList(variable="__parents", recurse_on='parents', recursion_depth=0, items=[
|
|
FormatAstVariable("__item", None, r111, 0)])
|
|
]),
|
|
FormatAstVariable("__item", None, r2, 1),
|
|
FormatAstList(variable="__parents", recurse_on='parents', recursion_depth=1, items=[
|
|
FormatAstVariable("__item", None, r22, 0)]),
|
|
])
|
|
|
|
def test_i_can_develop_using_variable_properties(self):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(
|
|
("isinstance(__obj, Concept)", "{__obj.body}"),
|
|
)
|
|
lst = Concept("bar", body="value for bar").auto_init()
|
|
|
|
res = service.create_out_tree(context, lst)
|
|
assert res == FormatAstVariable("__obj.body", None, "value for bar")
|
|
|
|
@pytest.mark.parametrize("rule, expected", [
|
|
(("__ret", "red('hello world')"), FormatAstColor("red", FormatAstRawText("hello world"))),
|
|
(("__ret", "blue(__ret_value)"), FormatAstColor("blue", FormatAstVariable("__ret_value", None, 1))),
|
|
])
|
|
def test_i_can_develop_color(self, rule, expected):
|
|
sheerka, context, service, rule = self.init_service_with_rules(rule)
|
|
ret = sheerka.ret("Test", True, 1)
|
|
|
|
res = service.create_out_tree(context, ret)
|
|
|
|
assert res == expected
|
|
|
|
@pytest.mark.parametrize("rule, expected", [
|
|
(("__ret", "{__ret}"), FormatAstVariable("__ret", None,
|
|
ReturnValueConcept(who="Test", status=True, value=1))),
|
|
(("__ret", "magenta(__ret)"),
|
|
FormatAstColor("magenta",
|
|
FormatAstVariable("__ret", None,
|
|
ReturnValueConcept(who="Test", status=True, value=1)))),
|
|
])
|
|
def test_i_can_manage_infinite_recursion(self, rule, expected):
|
|
sheerka, context, service, rule = self.init_service_with_rules(rule)
|
|
ret = sheerka.ret("Test", True, 1)
|
|
|
|
res = service.create_out_tree(context, ret)
|
|
|
|
assert res == expected
|
|
|
|
def test_i_can_manage_infinite_recursion_when_multiple_rules(self):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(
|
|
Rule("print", None, "__ret", "{__ret.value}", 1),
|
|
Rule("print", None, "__ret", "white(__ret)", 2),
|
|
)
|
|
ret = sheerka.ret("Test", True, "hello world!")
|
|
|
|
res = service.create_out_tree(context, ret)
|
|
|
|
assert res == FormatAstColor("white",
|
|
FormatAstVariable("__ret", None,
|
|
FormatAstVariable("__ret.value", None, "hello world!")))
|
|
|
|
def test_i_can_print_out_the_result(self, capsys):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(
|
|
("__rets", "list(__rets)"),
|
|
("__ret", "status: {status}, value: {__ret_value}"),
|
|
("isinstance(__obj, 'foo')", "{id}-{name}:{body}")
|
|
)
|
|
sheerka.create_new_concept(context, Concept("foo"))
|
|
foo = sheerka.new("foo", body="hello world")
|
|
ret = sheerka.ret("Test", True, foo)
|
|
|
|
service.process_return_values(context, [ret])
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "status: True, value: 1001-foo:hello world\n"
|
|
|
|
def test_i_can_print_out_a_list_of_results(self, capsys):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(
|
|
("__rets", "list(__rets)"),
|
|
("isinstance(__ret_value, 'foo')", "foo: {body}"),
|
|
("isinstance(__ret_value, 'bar')", "bar: {body}"),
|
|
)
|
|
sheerka.create_new_concept(context, Concept("foo"))
|
|
sheerka.create_new_concept(context, Concept("bar"))
|
|
sheerka.create_new_concept(context, Concept("baz"))
|
|
|
|
rets = [sheerka.ret("Test", True, sheerka.new("foo", body="foo value")),
|
|
sheerka.ret("Test", True, sheerka.new("bar", body="bar value")),
|
|
sheerka.ret("Test", True, sheerka.new("baz", body="baz value")), ]
|
|
|
|
service.process_return_values(context, rets)
|
|
captured = capsys.readouterr()
|
|
assert captured.out == """foo: (1001)foo
|
|
bar: (1002)bar
|
|
ReturnValue(who=Test, status=True, value=(1003)baz, message=None)
|
|
"""
|
|
|
|
def test_i_can_print_out_a_list_with_recurse(self, capsys):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(
|
|
("__rets", "list(__rets, recurse_on='parents', recursion_depth=2)"),
|
|
("__ret", "{__tab}{__ret}"),
|
|
)
|
|
r1111 = ReturnValueConcept("Test", True, "r1111")
|
|
r111 = ReturnValueConcept("Test", True, "r111", parents=[r1111])
|
|
r11 = ReturnValueConcept("Test", True, "r11", parents=[r111])
|
|
r1 = ReturnValueConcept("Test", True, "r1", parents=[r11])
|
|
r22 = ReturnValueConcept("Test", True, "r22")
|
|
r2 = ReturnValueConcept("Test", True, "r2", parents=[r22])
|
|
|
|
rets = [r1, r2]
|
|
|
|
service.process_return_values(context, rets)
|
|
captured = capsys.readouterr()
|
|
assert captured.out == """ReturnValue(who=Test, status=True, value=r1, message=None)
|
|
ReturnValue(who=Test, status=True, value=r11, message=None)
|
|
ReturnValue(who=Test, status=True, value=r111, message=None)
|
|
ReturnValue(who=Test, status=True, value=r2, message=None)
|
|
ReturnValue(who=Test, status=True, value=r22, message=None)
|
|
"""
|
|
|
|
def test_i_can_print_out_a_list_with_recurse_using_format_instr(self, capsys):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(
|
|
("__rets", "list(__rets)"),
|
|
("__ret", "{__tab}{__ret}"),
|
|
)
|
|
r1111 = ReturnValueConcept("Test", True, "r1111")
|
|
r111 = ReturnValueConcept("Test", True, "r111", parents=[r1111])
|
|
r11 = ReturnValueConcept("Test", True, "r11", parents=[r111])
|
|
r1 = ReturnValueConcept("Test", True, "r1", parents=[r11])
|
|
r1.set_format_instr(recursion_depth=2, recurse_on="parents")
|
|
r22 = ReturnValueConcept("Test", True, "r22")
|
|
r2 = ReturnValueConcept("Test", True, "r2", parents=[r22])
|
|
r2.set_format_instr(recursion_depth=2, recurse_on="parents")
|
|
|
|
rets = [r1, r2]
|
|
|
|
service.process_return_values(context, rets)
|
|
captured = capsys.readouterr()
|
|
assert captured.out == """ReturnValue(who=Test, status=True, value=r1, message=None)
|
|
ReturnValue(who=Test, status=True, value=r11, message=None)
|
|
ReturnValue(who=Test, status=True, value=r111, message=None)
|
|
ReturnValue(who=Test, status=True, value=r2, message=None)
|
|
ReturnValue(who=Test, status=True, value=r22, message=None)
|
|
"""
|
|
|
|
def test_i_can_print_out_a_list_with_recurse_using_container_format_instr(self, capsys):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(
|
|
("isinstance(__obj, 'foo')", "list(__obj)"),
|
|
("__ret", "{__tab}{__ret}"),
|
|
)
|
|
r1111 = ReturnValueConcept("Test", True, "r1111")
|
|
r111 = ReturnValueConcept("Test", True, "r111", parents=[r1111])
|
|
r11 = ReturnValueConcept("Test", True, "r11", parents=[r111])
|
|
r1 = ReturnValueConcept("Test", True, "r1", parents=[r11])
|
|
r22 = ReturnValueConcept("Test", True, "r22")
|
|
r2 = ReturnValueConcept("Test", True, "r2", parents=[r22])
|
|
|
|
foo = Concept("foo", key="foo", body=[r1, r2]).auto_init()
|
|
foo.set_format_instr(recurse_on='parents', recursion_depth=2)
|
|
|
|
service.process_return_values(context, foo)
|
|
captured = capsys.readouterr()
|
|
assert captured.out == """ReturnValue(who=Test, status=True, value=r1, message=None)
|
|
ReturnValue(who=Test, status=True, value=r11, message=None)
|
|
ReturnValue(who=Test, status=True, value=r111, message=None)
|
|
ReturnValue(who=Test, status=True, value=r2, message=None)
|
|
ReturnValue(who=Test, status=True, value=r22, message=None)
|
|
"""
|
|
|
|
def test_i_can_print_out_color(self, capsys):
|
|
sheerka, context, service, *rules = self.init_service_with_rules(
|
|
("__ret", "status: green(status), blue('value: {__ret_value}')"),
|
|
)
|
|
sheerka.create_new_concept(context, Concept("foo"))
|
|
foo = sheerka.new("foo", body="hello world")
|
|
ret = sheerka.ret("Test", True, foo)
|
|
|
|
service.process_return_values(context, ret)
|
|
captured = capsys.readouterr()
|
|
assert captured.out == "status: \x1b[32mTrue\x1b[0m, \x1b[34mvalue: (1001)foo\x1b[0m\n"
|