sartography
/
spiffworkflow-backend
mirror of https://github.com/sartography/spiffworkflow-backend.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

improve ScriptUnitTestRunner API and tests

This commit is contained in:
burnettk 2022-09-19 18:38:21 -04:00
parent 54dac63b1e
commit 4efe56f46f
4 changed files with 213 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
src/spiffworkflow_backend/services/script_unit_test_runner.py
View File

@ -1,9 +1,12 @@
"""Process_instance_processor."""
from dataclasses import dataclass
from typing import Any
from SpiffWorkflow import Task as SpiffTask # type: ignore
from spiffworkflow_backend.services.process_instance_processor import CustomBpmnScriptEngine
from spiffworkflow_backend.services.process_instance_processor import (
CustomBpmnScriptEngine,
)
PythonScriptContext = dict[str, Any]
@ -11,6 +14,7 @@ PythonScriptContext = dict[str, Any]
@dataclass
class ScriptUnitTestResult:
"""ScriptUnitTestResult."""
result: bool
context: PythonScriptContext
@ -21,13 +25,56 @@ class ScriptUnitTestRunner:
_script_engine = CustomBpmnScriptEngine()
@classmethod
def run_task(
cls, task: SpiffTask, script: str, input_context: PythonScriptContext, expected_output_context: PythonScriptContext
def run_with_task_and_script_and_pre_post_contexts(
cls,
task: SpiffTask,
script: str,
input_context: PythonScriptContext,
expected_output_context: PythonScriptContext,
) -> ScriptUnitTestResult:
"""Run_task."""
task.data = input_context
cls._script_engine.execute(task, script)
result_as_boolean = task.data == expected_output_context
script_unit_test_result = ScriptUnitTestResult(result=result_as_boolean, context=task.data)
script_unit_test_result = ScriptUnitTestResult(
result=result_as_boolean, context=task.data
)
return script_unit_test_result
# run_test_temp is just so we can write tests against this class until run_test (below) is implementable
# when spiffworkflow starts exposing the unit tests of each script task in the spiff xml extensions
@classmethod
def run_test_temp(
cls,
task: SpiffTask,
input_context: PythonScriptContext,
expected_output_context: PythonScriptContext,
) -> ScriptUnitTestResult:
"""Run_test_temp."""
script = task.task_spec.script
return cls.run_with_task_and_script_and_pre_post_contexts(
task, script, input_context, expected_output_context
)
@classmethod
def run_test(
cls,
task: SpiffTask,
test_identifier: str,
) -> ScriptUnitTestResult:
"""Run_test."""
# this is totally made up, but hopefully resembles what spiffworkflow ultimately does
unit_tests = task.task_spec.extensions["unit_tests"]
unit_test = [
unit_test
for unit_test in unit_tests
if unit_test["test_identifier"] == test_identifier
][0]
input_context = unit_test["input_json"]
expected_output_context = unit_test["expected_output_json"]
script = task.task_spec.script
return cls.run_with_task_and_script_and_pre_post_contexts(
task, script, input_context, expected_output_context
)

69
tests/data/script_with_unit_tests/script_with_unit_tests.bpmn Normal file
View File

@ -0,0 +1,69 @@
<?xml version="1.0" encoding="UTF-8"?>
<bpmn:definitions xmlns:bpmn="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI" xmlns:dc="http://www.omg.org/spec/DD/20100524/DC" xmlns:di="http://www.omg.org/spec/DD/20100524/DI" xmlns:spiffworkflow="http://spiffworkflow.org/bpmn/schema/1.0/core" xmlns:modeler="http://camunda.org/schema/modeler/1.0" id="Definitions_1ny7jp4" targetNamespace="http://bpmn.io/schema/bpmn" exporter="Camunda Modeler" exporterVersion="5.0.0" modeler:executionPlatform="Camunda Platform" modeler:executionPlatformVersion="7.17.0">
<bpmn:process id="sample" isExecutable="true">
<bpmn:startEvent id="StartEvent_1">
<bpmn:outgoing>Flow_10jwwqy</bpmn:outgoing>
</bpmn:startEvent>
<bpmn:sequenceFlow id="Flow_10jwwqy" sourceRef="StartEvent_1" targetRef="Activity_03fldr6" />
<bpmn:endEvent id="Event_1qb1u6a">
<bpmn:incoming>Flow_0htxke7</bpmn:incoming>
</bpmn:endEvent>
<bpmn:sequenceFlow id="Flow_0htxke7" sourceRef="script_with_unit_test_id" targetRef="Event_1qb1u6a" />
<bpmn:scriptTask id="script_with_unit_test_id" name="Script with unit test">
<bpmn:extensionElements>
<spiffworkflow:unitTests>
<spiffworkflow:unitTest id="sets_hey_to_true_if_hey_is_false">
<spiffworkflow:inputJson>{"hey": false}</spiffworkflow:inputJson>
<spiffworkflow:expectedOutputJson>{"hey": true}</spiffworkflow:expectedOutputJson>
</spiffworkflow:unitTest>
<spiffworkflow:unitTest id="sets_something_else_if_no_hey">
<spiffworkflow:inputJson>{}</spiffworkflow:inputJson>
<spiffworkflow:expectedOutputJson>{"something_else": true}</spiffworkflow:expectedOutputJson>
</spiffworkflow:unitTest>
</spiffworkflow:unitTests>
</bpmn:extensionElements>
<bpmn:incoming>Flow_0niwe1y</bpmn:incoming>
<bpmn:outgoing>Flow_0htxke7</bpmn:outgoing>
<bpmn:script>if 'hey' in locals():
hey = True
else:
something_else = True</bpmn:script>
</bpmn:scriptTask>
<bpmn:sequenceFlow id="Flow_0niwe1y" sourceRef="Activity_03fldr6" targetRef="script_with_unit_test_id" />
<bpmn:scriptTask id="Activity_03fldr6" name="Set var">
<bpmn:incoming>Flow_10jwwqy</bpmn:incoming>
<bpmn:outgoing>Flow_0niwe1y</bpmn:outgoing>
<bpmn:script>hey = False</bpmn:script>
</bpmn:scriptTask>
</bpmn:process>
<bpmndi:BPMNDiagram id="BPMNDiagram_1">
<bpmndi:BPMNPlane id="BPMNPlane_1" bpmnElement="sample">
<bpmndi:BPMNShape id="_BPMNShape_StartEvent_2" bpmnElement="StartEvent_1">
<dc:Bounds x="132" y="102" width="36" height="36" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Event_1qb1u6a_di" bpmnElement="Event_1qb1u6a">
<dc:Bounds x="642" y="102" width="36" height="36" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Activity_17ohe7r_di" bpmnElement="script_with_unit_test_id">
<dc:Bounds x="440" y="80" width="100" height="80" />
<bpmndi:BPMNLabel />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Activity_1fab1y2_di" bpmnElement="Activity_03fldr6">
<dc:Bounds x="250" y="80" width="100" height="80" />
<bpmndi:BPMNLabel />
</bpmndi:BPMNShape>
<bpmndi:BPMNEdge id="Flow_10jwwqy_di" bpmnElement="Flow_10jwwqy">
<di:waypoint x="168" y="120" />
<di:waypoint x="250" y="120" />
</bpmndi:BPMNEdge>
<bpmndi:BPMNEdge id="Flow_0htxke7_di" bpmnElement="Flow_0htxke7">
<di:waypoint x="540" y="120" />
<di:waypoint x="642" y="120" />
</bpmndi:BPMNEdge>
<bpmndi:BPMNEdge id="Flow_0niwe1y_di" bpmnElement="Flow_0niwe1y">
<di:waypoint x="350" y="120" />
<di:waypoint x="440" y="120" />
</bpmndi:BPMNEdge>
</bpmndi:BPMNPlane>
</bpmndi:BPMNDiagram>
</bpmn:definitions>

7
tests/spiffworkflow_backend/unit/test_process_instance_processor.py
View File

@ -1,6 +1,9 @@
"""Test_process_instance_processor."""
from flask.app import Flask
from spiffworkflow_backend.services.process_instance_processor import ProcessInstanceProcessor
from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor,
)
# it's not totally obvious we want to keep this test/file
@ -11,5 +14,5 @@ def test_script_engine_takes_data_and_returns_expected_results(
"""Test_script_engine_takes_data_and_returns_expected_results."""
script_engine = ProcessInstanceProcessor._script_engine
result = script_engine._evaluate('a', {"a": 1})
result = script_engine._evaluate("a", {"a": 1})
assert result == 1

93
tests/spiffworkflow_backend/unit/test_script_unit_test_runner.py
View File

@ -1,34 +1,44 @@
"""Test Permissions."""
from flask.app import Flask
from spiffworkflow_backend.services.process_instance_processor import ProcessInstanceProcessor
from spiffworkflow_backend.services.script_unit_test_runner import PythonScriptContext, ScriptUnitTestRunner
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend.services.process_instance_processor import (
ProcessInstanceProcessor,
)
from spiffworkflow_backend.services.script_unit_test_runner import PythonScriptContext
from spiffworkflow_backend.services.script_unit_test_runner import ScriptUnitTestRunner
class TestScriptUnitTestRunner(BaseTest):
"""TestScriptUnitTestRunner."""
def test_takes_data_and_returns_expected_result(self,
def test_takes_data_and_returns_expected_result(
self,
app: Flask,
with_db_and_bpmn_file_cleanup: None,
) -> None:
"""Test_takes_data_and_returns_expected_result."""
process_group_id = "test_logging_spiff_logger"
process_model_id = "simple_script"
process_model = load_test_spec(process_model_id, process_group_id=process_group_id)
process_model = load_test_spec(
process_model_id, process_group_id=process_group_id
)
process_instance = self.create_process_instance_from_process_model(
process_model
)
processor = ProcessInstanceProcessor(process_instance)
task = processor.get_task_by_bpmn_identifier('Activity_RunScript')
task = processor.get_task_by_bpmn_identifier("Activity_RunScript")
input_context: PythonScriptContext = {"a": 1}
expected_output_context: PythonScriptContext = {"a": 2}
script = "a = 2"
unit_test_result = ScriptUnitTestRunner.run_task(task, script, input_context, expected_output_context)
unit_test_result = (
ScriptUnitTestRunner.run_with_task_and_script_and_pre_post_contexts(
task, script, input_context, expected_output_context
)
)
assert unit_test_result.result
assert unit_test_result.context == {"a": 2}
@ -39,25 +49,86 @@ class TestScriptUnitTestRunner(BaseTest):
# result = script_engine._execute('a = 1', {})
# assert result == 1
def test_fails_when_expected_output_does_not_match_actual_output(self,
def test_fails_when_expected_output_does_not_match_actual_output(
self,
app: Flask,
with_db_and_bpmn_file_cleanup: None,
) -> None:
"""Test_fails_when_expected_output_does_not_match_actual_output."""
process_group_id = "test_logging_spiff_logger"
process_model_id = "simple_script"
process_model = load_test_spec(process_model_id, process_group_id=process_group_id)
process_model = load_test_spec(
process_model_id, process_group_id=process_group_id
)
process_instance = self.create_process_instance_from_process_model(
process_model
)
processor = ProcessInstanceProcessor(process_instance)
task = processor.get_task_by_bpmn_identifier('Activity_RunScript')
task = processor.get_task_by_bpmn_identifier("Activity_RunScript")
input_context: PythonScriptContext = {"a": 1}
expected_output_context: PythonScriptContext = {"a": 2, "b": 3}
script = "a = 2"
unit_test_result = ScriptUnitTestRunner.run_task(task, script, input_context, expected_output_context)
unit_test_result = (
ScriptUnitTestRunner.run_with_task_and_script_and_pre_post_contexts(
task, script, input_context, expected_output_context
)
)
assert unit_test_result.result is not True
assert unit_test_result.context == {"a": 2}
def test_script_with_unit_tests_when_hey_is_passed_in(
self,
app: Flask,
with_db_and_bpmn_file_cleanup: None,
) -> None:
"""Test_script_with_unit_tests_when_hey_is_passed_in."""
process_group_id = "script_with_unit_tests"
process_model_id = "script_with_unit_tests"
process_model = load_test_spec(
process_model_id, process_group_id=process_group_id
)
process_instance = self.create_process_instance_from_process_model(
process_model
)
processor = ProcessInstanceProcessor(process_instance)
task = processor.get_task_by_bpmn_identifier("script_with_unit_test_id")
input_context: PythonScriptContext = {"hey": 1}
expected_output_context: PythonScriptContext = {"hey": True}
unit_test_result = ScriptUnitTestRunner.run_test_temp(
task, input_context, expected_output_context
)
assert unit_test_result.result
assert unit_test_result.context == {"hey": True}
def test_script_with_unit_tests_when_hey_is_not_passed_in(
self,
app: Flask,
with_db_and_bpmn_file_cleanup: None,
) -> None:
"""Test_script_with_unit_tests_when_hey_is_not_passed_in."""
process_group_id = "script_with_unit_tests"
process_model_id = "script_with_unit_tests"
process_model = load_test_spec(
process_model_id, process_group_id=process_group_id
)
process_instance = self.create_process_instance_from_process_model(
process_model
)
processor = ProcessInstanceProcessor(process_instance)
task = processor.get_task_by_bpmn_identifier("script_with_unit_test_id")
input_context: PythonScriptContext = {}
expected_output_context: PythonScriptContext = {"something_else": True}
unit_test_result = ScriptUnitTestRunner.run_test_temp(
task, input_context, expected_output_context
)
assert unit_test_result.result
assert unit_test_result.context == expected_output_context