unblock bpmn unit testing with duplicate files, our script engine, get process initiator user (#1140)

Co-authored-by: burnettk <burnettk@users.noreply.github.com>
This commit is contained in:
Kevin Burnett 2024-02-29 16:21:13 -08:00 committed by GitHub
parent 107eaaff3e
commit 72629f5ed8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 27 additions and 2 deletions

View File

@ -366,6 +366,10 @@ class CustomBpmnScriptEngine(PythonScriptEngine): # type: ignore
if external_context: if external_context:
methods.update(external_context) methods.update(external_context)
if hasattr(self, "method_overrides"):
if self.method_overrides:
methods = {**methods, **self.method_overrides}
"""Evaluate the given expression, within the context of the given task and return the result.""" """Evaluate the given expression, within the context of the given task and return the result."""
try: try:
return super()._evaluate(expression, context, external_context=methods) return super()._evaluate(expression, context, external_context=methods)
@ -387,6 +391,11 @@ class CustomBpmnScriptEngine(PythonScriptEngine): # type: ignore
methods = self.__get_augment_methods(task) methods = self.__get_augment_methods(task)
if external_context: if external_context:
methods.update(external_context) methods.update(external_context)
if hasattr(self, "method_overrides"):
if self.method_overrides:
methods = {**methods, **self.method_overrides}
# do not run script if it is blank # do not run script if it is blank
if script: if script:
super().execute(task, script, methods) super().execute(task, script, methods)

View File

@ -14,6 +14,7 @@ from SpiffWorkflow.util.deep_merge import DeepMerge # type: ignore
from SpiffWorkflow.util.task import TaskState # type: ignore from SpiffWorkflow.util.task import TaskState # type: ignore
from spiffworkflow_backend.services.custom_parser import MyCustomParser from spiffworkflow_backend.services.custom_parser import MyCustomParser
from spiffworkflow_backend.services.process_instance_processor import ProcessInstanceProcessor
class UnrunnableTestCaseError(Exception): class UnrunnableTestCaseError(Exception):
@ -108,6 +109,10 @@ class ProcessModelTestRunnerMostlyPureSpiffDelegate(ProcessModelTestRunnerDelega
raise BpmnFileMissingExecutableProcessError(f"Executable process cannot be found in {bpmn_file}. Test cannot run.") raise BpmnFileMissingExecutableProcessError(f"Executable process cannot be found in {bpmn_file}. Test cannot run.")
all_related = self._find_related_bpmn_files(bpmn_file) all_related = self._find_related_bpmn_files(bpmn_file)
# get unique list of related files
all_related = list(set(all_related))
for related_file in all_related: for related_file in all_related:
self._add_bpmn_file_to_parser(parser, related_file) self._add_bpmn_file_to_parser(parser, related_file)
@ -117,6 +122,17 @@ class ProcessModelTestRunnerMostlyPureSpiffDelegate(ProcessModelTestRunnerDelega
bpmn_process_spec, bpmn_process_spec,
subprocess_specs=subprocesses, subprocess_specs=subprocesses,
) )
bpmn_process_instance.script_engine = ProcessInstanceProcessor._default_script_engine
# we do not want to call the real get_process_initiator_user script, since it depends on a process instance
# that does not actually exist
overridden_methods = {
"get_process_initiator_user": lambda: {
"username": "test_username_a",
"tenant_specific_field_1": "test_tenant_specific_field_1_a",
},
}
bpmn_process_instance.script_engine.method_overrides = overridden_methods
return bpmn_process_instance return bpmn_process_instance
def execute_task(self, spiff_task: SpiffTask, task_data_for_submit: dict | None = None) -> None: def execute_task(self, spiff_task: SpiffTask, task_data_for_submit: dict | None = None) -> None:
@ -444,7 +460,7 @@ class ProcessModelTestRunner:
return test_mappings return test_mappings
class ProcessModeltTestRunnerBackendDelegate(ProcessModelTestRunnerMostlyPureSpiffDelegate): class ProcessModelTestRunnerBackendDelegate(ProcessModelTestRunnerMostlyPureSpiffDelegate):
pass pass
@ -459,7 +475,7 @@ class ProcessModelTestRunnerService:
process_model_directory_path, process_model_directory_path,
test_case_file=test_case_file, test_case_file=test_case_file,
test_case_identifier=test_case_identifier, test_case_identifier=test_case_identifier,
process_model_test_runner_delegate_class=ProcessModeltTestRunnerBackendDelegate, process_model_test_runner_delegate_class=ProcessModelTestRunnerBackendDelegate,
) )
def run(self) -> None: def run(self) -> None: