222 lines
8.1 KiB
Python
222 lines
8.1 KiB
Python
import ast
|
|
|
|
from core.builtin_concepts import ReturnValueConcept, ParserResultConcept, BuiltinConcepts
|
|
from core.concept import Concept, DEFINITION_TYPE_BNF, DEFINITION_TYPE_DEF
|
|
from core.rule import Rule
|
|
from core.sheerka.ExecutionContext import ExecutionContext
|
|
from core.sheerka.services.SheerkaRuleManager import SheerkaRuleManager
|
|
from parsers.BnfDefinitionParser import BnfDefinitionParser
|
|
from parsers.BnfNodeParser import StrMatch
|
|
from sdp.sheerkaDataProvider import Event
|
|
|
|
|
|
class BaseTest:
|
|
def get_sheerka(self, **kwargs):
|
|
pass
|
|
|
|
def get_context(self, sheerka=None, eval_body=False, eval_where=False):
|
|
context = ExecutionContext("test", Event(), 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
|
|
|
|
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):
|
|
sheerka = self.get_sheerka(**kwargs)
|
|
|
|
context_args = dict([(k, v) for k, v in kwargs.items() if k in ["sheerka", "eval_body", "eval_where"]])
|
|
context = self.get_context(sheerka, **context_args)
|
|
create_new = kwargs.get("create_new", None)
|
|
|
|
result = []
|
|
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 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}'",
|
|
sheerka.get_error(res))
|
|
|
|
if create_new:
|
|
sheerka.create_new_concept(context, c)
|
|
else:
|
|
c.init_key()
|
|
sheerka.set_id_if_needed(c, False)
|
|
sheerka.add_in_cache(c)
|
|
|
|
result.append(c)
|
|
|
|
return sheerka, context, *result
|
|
|
|
def init_format_rules(self, *rules, create_new=True, compile_rule=True, concepts=None, **kwargs):
|
|
if concepts:
|
|
sheerka, context, *concepts = self.init_concepts(*concepts, **kwargs)
|
|
else:
|
|
sheerka, context, *concepts = self.init_concepts(**kwargs)
|
|
|
|
if create_new:
|
|
sheerka.cache_manager.caches[SheerkaRuleManager.FORMAT_RULE_ENTRY].cache.clear()
|
|
sheerka.cache_manager.delete(sheerka.CONCEPTS_KEYS_ENTRY, SheerkaRuleManager.RULE_IDS)
|
|
with sheerka.sdp.get_transaction(context.event.get_digest()) as transaction:
|
|
transaction.clear(SheerkaRuleManager.FORMAT_RULE_ENTRY)
|
|
|
|
initialized = []
|
|
for rule_blue_print in rules:
|
|
if isinstance(rule_blue_print, tuple):
|
|
rule = Rule("print", None, rule_blue_print[0], rule_blue_print[1])
|
|
if not compile_rule:
|
|
rule.metadata.is_compiled = True
|
|
else:
|
|
rule = rule_blue_print
|
|
|
|
is_enabled = rule.metadata.is_enabled
|
|
sheerka.services[SheerkaRuleManager.NAME].init_rule(context, rule)
|
|
if create_new:
|
|
res = sheerka.create_new_rule(context, rule)
|
|
initialized.append(res.body.body)
|
|
else:
|
|
initialized.append(rule)
|
|
|
|
if is_enabled is not None:
|
|
rule.metadata.is_enabled = is_enabled
|
|
|
|
return sheerka, context, *initialized
|
|
|
|
@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()
|
|
if sheerka.has_key(obj.key):
|
|
sheerka.add_in_cache(obj)
|
|
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.add_in_cache(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
|