bpmn unit test framework improvements for json issue (#1162)

* added the contents of script engine init to the process model test running init w/ burnettk

* get backend scripts

---------

Co-authored-by: jasquat <jasquat@users.noreply.github.com>
Co-authored-by: burnettk <burnettk@users.noreply.github.com>
This commit is contained in:
Kevin Burnett 2024-03-05 03:49:18 -08:00 committed by GitHub
parent 1167187557
commit bcf0593ebc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 59 additions and 9 deletions

View File

@ -1,13 +1,22 @@
import decimal
import glob
import json
import os
import re
import time
import traceback
import uuid
from abc import abstractmethod
from dataclasses import dataclass
from datetime import datetime
from datetime import timedelta
from typing import Any
import _strptime # type: ignore
import dateparser
import pytz
from lxml import etree # type: ignore
from RestrictedPython import safe_globals # type: ignore
from SpiffWorkflow.bpmn.exceptions import WorkflowTaskException # type: ignore
from SpiffWorkflow.bpmn.script_engine import PythonScriptEngine # type: ignore
from SpiffWorkflow.bpmn.workflow import BpmnWorkflow # type: ignore
@ -15,7 +24,11 @@ from SpiffWorkflow.task import Task as SpiffTask # type: ignore
from SpiffWorkflow.util.deep_merge import DeepMerge # type: ignore
from SpiffWorkflow.util.task import TaskState # type: ignore
from spiffworkflow_backend.models.script_attributes_context import ScriptAttributesContext
from spiffworkflow_backend.scripts.script import Script
from spiffworkflow_backend.services.custom_parser import MyCustomParser
from spiffworkflow_backend.services.jinja_service import JinjaHelpers
from spiffworkflow_backend.services.process_instance_processor import CustomScriptEngineEnvironment
class UnrunnableTestCaseError(Exception):
@ -42,12 +55,43 @@ class BpmnFileMissingExecutableProcessError(Exception):
pass
def _import(name: str, glbls: dict[str, Any], *args: Any) -> None:
if name not in glbls:
raise ImportError(f"Import not allowed: {name}", name=name)
class ProcessModelTestRunnerScriptEngine(PythonScriptEngine): # type: ignore
def __init__(self, method_overrides: dict | None = None) -> None:
self.method_overrides = method_overrides
super().__init__()
default_globals = {
"_strptime": _strptime,
"dateparser": dateparser,
"datetime": datetime,
"decimal": decimal,
"dict": dict,
"enumerate": enumerate,
"filter": filter,
"format": format,
"json": json,
"list": list,
"map": map,
"pytz": pytz,
"set": set,
"sum": sum,
"time": time,
"timedelta": timedelta,
"uuid": uuid,
**JinjaHelpers.get_helper_mapping(),
}
def _get_all_methods_for_context(self, external_context: dict[str, Any] | None) -> dict:
# This will overwrite the standard builtins
default_globals.update(safe_globals)
default_globals["__builtins__"]["__import__"] = _import
environment = CustomScriptEngineEnvironment(default_globals)
self.method_overrides = method_overrides
super().__init__(environment=environment)
def _get_all_methods_for_context(self, external_context: dict[str, Any] | None, task: SpiffTask | None = None) -> dict:
methods = {
"get_process_initiator_user": lambda: {
"username": "test_username_a",
@ -55,6 +99,14 @@ class ProcessModelTestRunnerScriptEngine(PythonScriptEngine): # type: ignore
},
}
script_attributes_context = ScriptAttributesContext(
task=task,
environment_identifier="mocked-environment-identifier",
process_instance_id=1,
process_model_identifier="fake-test-process-model-identifier",
)
methods = Script.generate_augmented_list(script_attributes_context)
if self.method_overrides:
methods = {**methods, **self.method_overrides}
@ -63,17 +115,15 @@ class ProcessModelTestRunnerScriptEngine(PythonScriptEngine): # type: ignore
return methods
# Evaluate the given expression, within the context of the given task and
# return the result.
def evaluate(self, task: SpiffTask, expression: str, external_context: dict[str, Any] | None = None) -> Any:
"""
Evaluate the given expression, within the context of the given task and
return the result.
"""
updated_context = self._get_all_methods_for_context(external_context)
updated_context = self._get_all_methods_for_context(external_context, task)
return super().evaluate(task, expression, updated_context)
def execute(self, task: SpiffTask, script: str, external_context: Any = None) -> bool:
if script:
methods = self._get_all_methods_for_context(external_context)
methods = self._get_all_methods_for_context(external_context, task)
super().execute(task, script, methods)
return True