from dataclasses import dataclass 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, FormatAstDict from core.utils import flatten_all_children from tests.TestUsingMemoryBasedSheerka import TestUsingMemoryBasedSheerka seq = FormatAstSequence raw = FormatAstRawText var = FormatAstVariable @dataclass class DummyObj: prop_1: float prop_2: str 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", value=1)), (("__ret", "{status}"), FormatAstVariable("status", value=True)), (("__ret", "{foo}"), FormatAstVariableNotFound("foo")), (("__ret", "hello world {__ret_value} !"), seq([FormatAstRawText("hello world "), FormatAstVariable("__ret_value", value=1), FormatAstRawText(" !")])), (("__ret", "red(__ret_value)"), FormatAstColor("red", FormatAstVariable("__ret_value", value=1))), (("__ret", "blue('hello world {__ret_value} !')"), FormatAstColor("blue", seq([FormatAstRawText("hello world "), FormatAstVariable("__ret_value", value=1), FormatAstRawText(" !")]))), (("__ret", "list(foo)"), FormatAstVariableNotFound("foo")), (("__ret", "{__ret_value:3}"), FormatAstVariable("__ret_value", "3", value=" 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", value=True), FormatAstRawText(", value: "), FormatAstVariable("__ret_value", value=seq([FormatAstVariable("id", value="1001"), FormatAstRawText("-"), FormatAstVariable("name", value="foo"), FormatAstRawText(":"), FormatAstVariable("body", value="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", value=True), FormatAstRawText(", value: "), FormatAstVariable("__ret.body", value=seq([FormatAstVariable("id", value="1001"), FormatAstRawText("-"), FormatAstVariable("name", value="foo"), FormatAstRawText(":"), FormatAstVariable("body", value="hello world")]))]) def test_i_can_develop_using_variable_properties(self): sheerka, context, service, *rules = self.init_service_with_rules( ("isinstance(__obj, Concept)", "{__obj.body}"), ) obj = Concept("bar", body="value for bar").auto_init() res = service.create_out_tree(context, obj) assert res == FormatAstVariable("__obj.body", value="value for bar") 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", value=lst[0], index=0), FormatAstVariable("__item", value=lst[1], index=1), FormatAstVariable("__item", value=lst[2], index=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", value=lst[0], index=0), FormatAstVariable("__item", value=lst[1], index=1), FormatAstVariable("__item", value=lst[2], index=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", value=lst[0], index=0), FormatAstVariable("__item", value=lst[1], index=1), FormatAstVariable("__item", value=lst[2], index=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", value=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", value=lst[0], index=0), FormatAstVariable("__item", value=lst[1], index=1), FormatAstVariable("__item", value=lst[2], index=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", value=FormatAstColor("red", FormatAstVariable("__ret", value=lst[0])), index=0), FormatAstVariable("__item", value=FormatAstColor("red", FormatAstVariable("__ret", value=lst[1])), index=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", value=r1, index=0), FormatAstList(variable="__parents", recurse_on='parents', recursion_depth=1, items=[ FormatAstVariable("__item", value=r11, index=0), FormatAstList(variable="__parents", recurse_on='parents', recursion_depth=0, items=[ FormatAstVariable("__item", value=r111, index=0)]) ]), FormatAstVariable("__item", value=r2, index=1), FormatAstList(variable="__parents", recurse_on='parents', recursion_depth=1, items=[ FormatAstVariable("__item", value=r22, index=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", value=r1, index=0), FormatAstList(variable="__parents", recurse_on='parents', recursion_depth=1, items=[ FormatAstVariable("__item", value=r11, index=0), FormatAstList(variable="__parents", recurse_on='parents', recursion_depth=0, items=[ FormatAstVariable("__item", value=r111, index=0)]) ]), FormatAstVariable("__item", value=r2, index=1), FormatAstList(variable="__parents", recurse_on='parents', recursion_depth=1, items=[ FormatAstVariable("__item", value=r22, index=0)]), ]) def test_i_can_develop_dict(self): sheerka, context, service, *rules = self.init_service_with_rules(("isinstance(__obj, dict)", "dict(__obj)")) obj = { "key1": "value1", "key2": "value2", } res = service.create_out_tree(context, obj) assert res == FormatAstDict(variable="__obj", items=[ (FormatAstVariable("__key", value="key1", index=0), FormatAstVariable("__value", value="value1", index="key1")), (FormatAstVariable("__key", value="key2", index=1), FormatAstVariable("__value", value="value2", index="key2")), ]) def test_i_can_develop_dict_with_list(self): sheerka, context, service, *rules = self.init_service_with_rules(("isinstance(__obj, dict)", "dict(__obj)")) obj = { "key1": "value1", "key2": ["item1", "item2"], } res = service.create_out_tree(context, obj) assert res == FormatAstDict(variable="__obj", items=[ (FormatAstVariable("__key", value="key1", index=0), FormatAstVariable("__value", value="value1", index="key1")), (FormatAstVariable("__key", value="key2", index=1), FormatAstList(variable='__value', debug=True, prefix='[', suffix=']', items=[ FormatAstVariable("__item", debug=True, value="item1", index=0), FormatAstVariable("__item", debug=True, value="item2", index=1), ])), ]) @pytest.mark.parametrize("rule, expected", [ (("__ret", "red('hello world')"), FormatAstColor("red", FormatAstRawText("hello world"))), (("__ret", "blue(__ret_value)"), FormatAstColor("blue", FormatAstVariable("__ret_value", value=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", value=ReturnValueConcept(who="Test", status=True, value=1))), (("__ret", "magenta(__ret)"), FormatAstColor("magenta", FormatAstVariable("__ret", value=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", value=FormatAstVariable("__ret.value", value="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" def test_i_can_print_out_dict(self, capsys): sheerka, context, service, *rules = self.init_service_with_rules( ("isinstance(__obj, dict)", "dict(__obj)"), ("__key=='key1'", "green(__key)") ) obj = { "key1": "value1", "key2": "value2", } service.process_return_values(context, obj) captured = capsys.readouterr() assert captured.out == """\x1b[32mkey1\x1b[0m: value1 key2: value2 """ def test_i_can_print_out_dict_in_debug_mode(self, capsys): sheerka, context, service, *rules = self.init_service_with_rules( ("isinstance(__obj, dict)", "dict(__obj, debug=True)"), ("__key=='key1'", "green(__key)") ) obj = { "key1": "value1", "key2": 1, "key3": DummyObj(3.15, "a string"), "key4": ["alpha", 0] } service.process_return_values(context, obj) captured = capsys.readouterr() assert captured.out == """{'\x1b[32mkey1\x1b[0m': 'value1', 'key2': 1, 'key3': DummyObj(prop_1=3.15, prop_2='a string'), 'key4': ['alpha', 0]} """