import ast from dataclasses import dataclass, field from core.builtin_concepts import ReturnValueConcept, ParserResultConcept, BuiltinConcepts from core.concept import Concept, DEFINITION_TYPE_BNF, DEFINITION_TYPE_DEF, freeze_concept_attrs from core.rule import Rule, ACTION_TYPE_PRINT, ACTION_TYPE_EXEC from core.sheerka.ExecutionContext import ExecutionContext from core.sheerka.Sheerka import Sheerka from core.sheerka.services.SheerkaConceptManager import SheerkaConceptManager from core.sheerka.services.SheerkaRuleManager import SheerkaRuleManager from parsers.BnfDefinitionParser import BnfDefinitionParser from parsers.BnfNodeParser import StrMatch from sdp.sheerkaDataProvider import Event @dataclass class InitTestHelper: sheerka: Sheerka context: ExecutionContext items: list = field(default_factory=list) def push(self, *items): self.items.extend(items) def unpack(self): return self.sheerka, self.context, *self.items def with_concepts(self, *concepts, **kwargs): create_new = kwargs.get("create_new", False) for c in concepts: if isinstance(c, str): c = Concept(c) if c.get_metadata().definition and c.get_metadata().definition_type != DEFINITION_TYPE_DEF: desc = f"Resolving BNF {c.get_metadata().definition}" with self.context.push(BuiltinConcepts.INIT_BNF, c, obj=c, desc=desc) as sub_context: bnf_parser = BnfDefinitionParser() res = bnf_parser.parse(sub_context, c.get_metadata().definition) if res.status: c.set_bnf(res.value.value) c.get_metadata().definition_type = DEFINITION_TYPE_BNF else: raise Exception(f"Error in bnf definition '{c.get_metadata().definition}'", self.sheerka.get_errors(self.context, res)) self._update_concept_parameters(c) if create_new: self.sheerka.create_new_concept(self.context, c) else: c.init_key() self.sheerka.set_id_if_needed(c, False) self.sheerka.test_only_add_in_cache(c) freeze_concept_attrs(c) self.items.append(c) return self def with_format_rules(self, *rules, **kwargs): return self.with_rules(ACTION_TYPE_PRINT, *rules, **kwargs) def with_exec_rules(self, *rules, **kwargs): return self.with_rules(ACTION_TYPE_EXEC, *rules, **kwargs) def with_rules(self, action_type, *rules, **kwargs): create_new = kwargs.get("create_new", True) compile_rule = kwargs.get("compile_rule", True) for rule_template in rules: if isinstance(rule_template, tuple): if len(rule_template) == 3: rule = Rule(action_type, rule_template[0], rule_template[1], rule_template[2]) else: rule = Rule(action_type, None, rule_template[0], rule_template[1]) else: rule = rule_template is_enabled = rule.metadata.is_enabled # remember the value... if compile_rule: self.sheerka.services[SheerkaRuleManager.NAME].init_rule(self.context, rule) else: rule.metadata.is_compiled = True if create_new: res = self.sheerka.create_new_rule(self.context, rule) if not res.status: raise Exception(f"Error in rule definition '{res.body}'", self.sheerka.get_errors(res)) self.items.append(res.body.body) else: self.items.append(rule) if is_enabled is not None: # ...and back the value if it was not None rule.metadata.is_enabled = is_enabled return self def _update_concept_parameters(self, concept): if concept.get_metadata().parameters: return SheerkaConceptManager.recompute_concept_parameters(self.context, concept) class BaseTest: def get_sheerka(self, **kwargs) -> Sheerka: pass def get_context(self, sheerka=None, eval_body=False, eval_where=False, global_truth=False, message=""): context = ExecutionContext("test", Event(message=message), sheerka or self.get_sheerka(), BuiltinConcepts.TESTING, None) if eval_body: context.protected_hints.add(BuiltinConcepts.EVAL_BODY_REQUESTED) if eval_where: context.protected_hints.add(BuiltinConcepts.EVAL_WHERE_REQUESTED) if global_truth: context.protected_hints.add(BuiltinConcepts.EVAL_GLOBAL_TRUTH_REQUESTED) return context @staticmethod def get_init_test_args(**kwargs): return {k: v for k, v in kwargs.items() if k in ["cache_only", "ontology", "eval_body", "eval_where", "global_truth"]} @staticmethod def get_with_concepts_args(**kwargs): return {k: v for k, v in kwargs.items() if k in ["create_new"]} def init_test(self, cache_only=None, ontology=None, eval_body=False, eval_where=False, global_truth=False): sheerka = self.get_sheerka(cache_only=cache_only, ontology=ontology) context = self.get_context(sheerka=sheerka, eval_body=eval_body, eval_where=eval_where, global_truth=global_truth) return InitTestHelper(sheerka, context) def get_default_concept(self): concept = Concept( name="a + b", where="isinstance(a, int) and isinstance(b, int)\n", pre="isinstance(a, int) and isinstance(b, int)\n", post="isinstance(res, int)\n", body="def func(x,y):\n return x+y\nfunc(a,b)", desc="specific description") concept.def_var("a", "value1") concept.def_var("b", "value2") return concept def dump_ast(self, node): dump = ast.dump(node) for to_remove in [", ctx=Load()", ", kind=None", ", type_ignores=[]"]: dump = dump.replace(to_remove, "") return dump @staticmethod def dump_tokens(tokens): return [t.repr_value for t in tokens] def init_concepts(self, *concepts, **kwargs): init_test_args = self.get_init_test_args(**kwargs) with_concepts_args = self.get_with_concepts_args(**kwargs) return self.init_test(**init_test_args).with_concepts(*concepts, **with_concepts_args).unpack() def init_format_rules(self, *rules, **kwargs): return self.init_test(**kwargs).with_format_rules(*rules, **kwargs).unpack() def init_exec_rules(self, *rules, **kwargs): return self.init_test(**kwargs).with_exec_rules(*rules, **kwargs).unpack() @staticmethod def get_concept_instance(sheerka, concept, **kwargs): """ Use to instantiate concept with default variables already set :param sheerka: :param concept: :param kwargs: :return: """ instance = sheerka.new(concept.key if isinstance(concept, Concept) else concept) for i, var in enumerate(instance.get_metadata().variables): if var[0] in kwargs: assert isinstance(kwargs[var[0]], str), "variables definitions must be string" instance.get_metadata().variables[i] = (var[0], kwargs[var[0]]) return instance @staticmethod def retval(obj, who="who", status=True): """ret_val""" return ReturnValueConcept(who, status, obj) @staticmethod def tretval(sheerka, obj, who="who"): """True ret_val + add concept in cache""" if isinstance(obj, Concept): obj.init_key() return sheerka.ret(who, True, obj) @staticmethod def pretval(concept, source=None, parser="parsers.name", who=None, status=True): """ParserResult ret_val (p stands for ParserResult)""" return ReturnValueConcept( who or parser, status, ParserResultConcept(parser=parser, source=source or concept.name, value=concept, try_parsed=concept)) @staticmethod def create_and_add_in_cache_concept(sheerka, name, variables=None, bnf=None): """ Create a concept using parameters and add it in cache :param sheerka: :param name: :param variables: :param bnf: :return: """ concept = Concept(name) if isinstance(name, str) else name if variables: for v in variables: concept.def_var(v) if bnf: concept.set_bnf(bnf) concept.get_metadata().definition_type = DEFINITION_TYPE_BNF concept.init_key() sheerka.set_id_if_needed(concept, False) sheerka.test_only_add_in_cache(concept) freeze_concept_attrs(concept) return concept @staticmethod def bnf_concept(concept, expression=None): if isinstance(concept, Concept): name = concept.name else: name = concept concept = Concept(concept) concept.set_bnf(expression or StrMatch(name)) concept.get_metadata().definition_type = DEFINITION_TYPE_BNF return concept @staticmethod def from_def_concept(name, definition, variables=None, **kwargs): concept = Concept(name=name, definition=definition, definition_type=DEFINITION_TYPE_DEF) if variables: for v in variables: concept.def_var(v) if kwargs: for k, v in kwargs.items(): if k in ("body", "pre", "post", "where"): setattr(concept.get_metadata(), k, v) else: concept.get_metadata().variables[k] = v return concept def init_scenario(self, init_expressions, global_truth=False): sheerka = self.get_sheerka() if global_truth: sheerka.add_to_context(BuiltinConcepts.EVAL_GLOBAL_TRUTH_REQUESTED) for expression in init_expressions: res = sheerka.evaluate_user_input(expression) assert len(res) == 1, f"Failed to execute '{expression}'" assert res[0].status, f"Error while executing '{expression}'" if global_truth: sheerka.remove_fom_context(BuiltinConcepts.EVAL_GLOBAL_TRUTH_REQUESTED) return sheerka @staticmethod def successful_return_values(return_values): return [ret_val for ret_val in return_values if ret_val.status]