Sheerka-Old/tests/BaseTest.py

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.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_error(res))

            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_error(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


class BaseTest:
    def get_sheerka(self, **kwargs) -> Sheerka:
        pass

    def get_context(self, sheerka=None, eval_body=False, eval_where=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)

        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"]}

    @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):
        sheerka = self.get_sheerka(cache_only=cache_only, ontology=ontology)
        context = self.get_context(sheerka=sheerka, eval_body=eval_body, eval_where=eval_where)

        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):
        sheerka = self.get_sheerka()

        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}'"

        return sheerka