do not require task to be given to evaluate a task unless that script specifically needs it w/ burnettk
This commit is contained in:
parent
afec26fbd5
commit
90b1772215
|
@ -1,5 +1,6 @@
|
|||
"""Script_attributes_context."""
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional
|
||||
|
||||
from SpiffWorkflow.task import Task as SpiffTask # type: ignore
|
||||
|
||||
|
@ -8,7 +9,7 @@ from SpiffWorkflow.task import Task as SpiffTask # type: ignore
|
|||
class ScriptAttributesContext:
|
||||
"""ScriptAttributesContext."""
|
||||
|
||||
task: SpiffTask
|
||||
task: Optional[SpiffTask]
|
||||
environment_identifier: str
|
||||
process_instance_id: int
|
||||
process_model_identifier: str
|
||||
|
|
|
@ -10,6 +10,10 @@ from spiffworkflow_backend.services.process_instance_processor import (
|
|||
)
|
||||
|
||||
|
||||
class TaskNotGivenToScriptError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class GetDataSizes(Script):
|
||||
"""GetDataSizes."""
|
||||
|
||||
|
@ -30,6 +34,11 @@ class GetDataSizes(Script):
|
|||
**kwargs: Any
|
||||
) -> Any:
|
||||
"""Run."""
|
||||
if script_attributes_context.task is None:
|
||||
raise TaskNotGivenToScriptError(
|
||||
"The task was not given to script 'get_data_sizes'. "
|
||||
"This script needs to be run from within the context of a task."
|
||||
)
|
||||
workflow = script_attributes_context.task.workflow
|
||||
task_data_size = ProcessInstanceProcessor.get_task_data_size(workflow)
|
||||
task_data_keys_by_task = {
|
||||
|
|
|
@ -316,7 +316,7 @@ class CustomBpmnScriptEngine(PythonScriptEngine): # type: ignore
|
|||
|
||||
super().__init__(environment=environment)
|
||||
|
||||
def __get_augment_methods(self, task: SpiffTask) -> Dict[str, Callable]:
|
||||
def __get_augment_methods(self, task: Optional[SpiffTask]) -> Dict[str, Callable]:
|
||||
"""__get_augment_methods."""
|
||||
tld = current_app.config["THREAD_LOCAL_DATA"]
|
||||
|
||||
|
@ -356,9 +356,7 @@ class CustomBpmnScriptEngine(PythonScriptEngine): # type: ignore
|
|||
external_methods: Optional[Dict[str, Any]] = None,
|
||||
) -> Any:
|
||||
"""_evaluate."""
|
||||
methods = {}
|
||||
if task:
|
||||
methods = self.__get_augment_methods(task)
|
||||
methods = self.__get_augment_methods(task)
|
||||
if external_methods:
|
||||
methods.update(external_methods)
|
||||
|
||||
|
|
|
@ -0,0 +1,28 @@
|
|||
"""Test_environment_var_script."""
|
||||
from flask import Flask
|
||||
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
|
||||
|
||||
from spiffworkflow_backend.models.script_attributes_context import (
|
||||
ScriptAttributesContext,
|
||||
)
|
||||
from spiffworkflow_backend.scripts.get_env import GetEnv
|
||||
|
||||
|
||||
class TestGetEnv(BaseTest):
|
||||
def test_get_env_script(
|
||||
self,
|
||||
app: Flask,
|
||||
with_db_and_bpmn_file_cleanup: None,
|
||||
) -> None:
|
||||
process_model_identifier = "test_process_model"
|
||||
process_instance_id = 1
|
||||
script_attributes_context = ScriptAttributesContext(
|
||||
task=None,
|
||||
environment_identifier="unit_testing",
|
||||
process_instance_id=process_instance_id,
|
||||
process_model_identifier=process_model_identifier,
|
||||
)
|
||||
result = GetEnv().run(
|
||||
script_attributes_context,
|
||||
)
|
||||
assert result == "unit_testing"
|
|
@ -1,23 +0,0 @@
|
|||
"""Test_environment_var_script."""
|
||||
from flask import Flask
|
||||
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
|
||||
|
||||
from spiffworkflow_backend.services.process_instance_processor import (
|
||||
ProcessInstanceProcessor,
|
||||
)
|
||||
|
||||
|
||||
class TestEnvironmentVarScript(BaseTest):
|
||||
"""TestEnvironmentVarScript."""
|
||||
|
||||
# it's not totally obvious we want to keep this test/file
|
||||
def test_script_engine_can_use_custom_scripts(
|
||||
self,
|
||||
app: Flask,
|
||||
with_db_and_bpmn_file_cleanup: None,
|
||||
) -> None:
|
||||
"""Test_script_engine_takes_data_and_returns_expected_results."""
|
||||
with app.app_context():
|
||||
script_engine = ProcessInstanceProcessor._script_engine
|
||||
result = script_engine._evaluate("get_env()", {})
|
||||
assert result == "unit_testing"
|
Loading…
Reference in New Issue