Added `get_localtime` script
Added test for new script, with test process
This commit is contained in:
mike cullerton
parent
6c1660a33d
commit
3471ab792a
|
|
@ -0,0 +1,42 @@
|
|||
from flask_bpmn.api.api_error import ApiError
|
||||
|
||||
from spiffworkflow_backend.scripts.script import Script
|
||||
|
||||
import pytz
|
||||
|
||||
from typing import Any
|
||||
from typing import Optional
|
||||
|
||||
from SpiffWorkflow.task import Task as SpiffTask # type: ignore
|
||||
|
||||
|
||||
class GetLocaltime(Script):
|
||||
|
||||
def get_description(self):
|
||||
return """Converts a Datetime object into a Datetime object for a specific timezone.
|
||||
Defaults to US/Eastern"""
|
||||
|
||||
def run(
|
||||
self,
|
||||
task: Optional[SpiffTask],
|
||||
environment_identifier: str,
|
||||
*args: Any,
|
||||
**kwargs: Any
|
||||
):
|
||||
if len(args) > 0 or 'datetime' in kwargs:
|
||||
if 'datetime' in kwargs:
|
||||
datetime = kwargs['datetime']
|
||||
else:
|
||||
datetime = args[0]
|
||||
if 'timezone' in kwargs:
|
||||
timezone = kwargs['timezone']
|
||||
elif len(args) > 1:
|
||||
timezone = args[1]
|
||||
else:
|
||||
timezone = 'US/Eastern'
|
||||
localtime = datetime.astimezone(pytz.timezone(timezone))
|
||||
return localtime
|
||||
|
||||
else:
|
||||
raise ApiError(status_code='missing_datetime',
|
||||
message='You must include a datetime to convert.')
|
||||
|
|
@ -0,0 +1,79 @@
|
|||
<?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:spiffworkflow="http://spiffworkflow.org/bpmn/schema/1.0/core" xmlns:di="http://www.omg.org/spec/DD/20100524/DI" id="Definitions_96f6665" targetNamespace="http://bpmn.io/schema/bpmn" exporter="Camunda Modeler" exporterVersion="3.0.0-dev">
|
||||
<bpmn:process id="Proccess_LocalTime" name="Get LocalTime" isExecutable="true">
|
||||
<bpmn:startEvent id="StartEvent_1">
|
||||
<bpmn:outgoing>Flow_0ijucqh</bpmn:outgoing>
|
||||
</bpmn:startEvent>
|
||||
<bpmn:sequenceFlow id="Flow_0ijucqh" sourceRef="StartEvent_1" targetRef="Activity_GetTimezone" />
|
||||
<bpmn:scriptTask id="Activity_GetTime" name="Get Time">
|
||||
<bpmn:documentation>Get Time</bpmn:documentation>
|
||||
<bpmn:incoming>Flow_10y2eax</bpmn:incoming>
|
||||
<bpmn:outgoing>Flow_1gcfk27</bpmn:outgoing>
|
||||
<bpmn:script>some_time = datetime.now()
|
||||
localtime = get_localtime(some_time, timezone)</bpmn:script>
|
||||
</bpmn:scriptTask>
|
||||
<bpmn:sequenceFlow id="Flow_1gcfk27" sourceRef="Activity_GetTime" targetRef="Activity_DisplayTime" />
|
||||
<bpmn:endEvent id="Event_1c50ix7">
|
||||
<bpmn:incoming>Flow_0dcc306</bpmn:incoming>
|
||||
</bpmn:endEvent>
|
||||
<bpmn:sequenceFlow id="Flow_0dcc306" sourceRef="Activity_DisplayTime" targetRef="Event_1c50ix7" />
|
||||
<bpmn:manualTask id="Activity_DisplayTime" name="Display Time">
|
||||
<bpmn:documentation>Display the time</bpmn:documentation>
|
||||
<bpmn:extensionElements>
|
||||
<spiffworkflow:instructionsForEndUser>## Time
|
||||
### Some Time: {{ some_time }}
|
||||
### Timezone: {{ timezone }}
|
||||
### Localtime: {{ localtime }}</spiffworkflow:instructionsForEndUser>
|
||||
</bpmn:extensionElements>
|
||||
<bpmn:incoming>Flow_1gcfk27</bpmn:incoming>
|
||||
<bpmn:outgoing>Flow_0dcc306</bpmn:outgoing>
|
||||
</bpmn:manualTask>
|
||||
<bpmn:sequenceFlow id="Flow_10y2eax" sourceRef="Activity_GetTimezone" targetRef="Activity_GetTime" />
|
||||
<bpmn:userTask id="Activity_GetTimezone" name="Get Timezone">
|
||||
<bpmn:extensionElements>
|
||||
<spiffworkflow:properties>
|
||||
<spiffworkflow:property name="formJsonSchemaFilename" value="get_localtime.json" />
|
||||
</spiffworkflow:properties>
|
||||
</bpmn:extensionElements>
|
||||
<bpmn:incoming>Flow_0ijucqh</bpmn:incoming>
|
||||
<bpmn:outgoing>Flow_10y2eax</bpmn:outgoing>
|
||||
</bpmn:userTask>
|
||||
</bpmn:process>
|
||||
<bpmndi:BPMNDiagram id="BPMNDiagram_1">
|
||||
<bpmndi:BPMNPlane id="BPMNPlane_1" bpmnElement="Proccess_LocalTime">
|
||||
<bpmndi:BPMNShape id="_BPMNShape_StartEvent_2" bpmnElement="StartEvent_1">
|
||||
<dc:Bounds x="179" y="159" width="36" height="36" />
|
||||
</bpmndi:BPMNShape>
|
||||
<bpmndi:BPMNShape id="Activity_1ahvmya_di" bpmnElement="Activity_GetTime">
|
||||
<dc:Bounds x="430" y="137" width="100" height="80" />
|
||||
<bpmndi:BPMNLabel />
|
||||
</bpmndi:BPMNShape>
|
||||
<bpmndi:BPMNShape id="Event_1c50ix7_di" bpmnElement="Event_1c50ix7">
|
||||
<dc:Bounds x="752" y="159" width="36" height="36" />
|
||||
</bpmndi:BPMNShape>
|
||||
<bpmndi:BPMNShape id="Activity_1gfsyxd_di" bpmnElement="Activity_DisplayTime">
|
||||
<dc:Bounds x="590" y="137" width="100" height="80" />
|
||||
</bpmndi:BPMNShape>
|
||||
<bpmndi:BPMNShape id="Activity_1x3u69s_di" bpmnElement="Activity_GetTimezone">
|
||||
<dc:Bounds x="270" y="137" width="100" height="80" />
|
||||
<bpmndi:BPMNLabel />
|
||||
</bpmndi:BPMNShape>
|
||||
<bpmndi:BPMNEdge id="Flow_0ijucqh_di" bpmnElement="Flow_0ijucqh">
|
||||
<di:waypoint x="215" y="177" />
|
||||
<di:waypoint x="270" y="177" />
|
||||
</bpmndi:BPMNEdge>
|
||||
<bpmndi:BPMNEdge id="Flow_1gcfk27_di" bpmnElement="Flow_1gcfk27">
|
||||
<di:waypoint x="530" y="177" />
|
||||
<di:waypoint x="590" y="177" />
|
||||
</bpmndi:BPMNEdge>
|
||||
<bpmndi:BPMNEdge id="Flow_0dcc306_di" bpmnElement="Flow_0dcc306">
|
||||
<di:waypoint x="690" y="177" />
|
||||
<di:waypoint x="752" y="177" />
|
||||
</bpmndi:BPMNEdge>
|
||||
<bpmndi:BPMNEdge id="Flow_10y2eax_di" bpmnElement="Flow_10y2eax">
|
||||
<di:waypoint x="370" y="177" />
|
||||
<di:waypoint x="430" y="177" />
|
||||
</bpmndi:BPMNEdge>
|
||||
</bpmndi:BPMNPlane>
|
||||
</bpmndi:BPMNDiagram>
|
||||
</bpmn:definitions>
|
||||
|
|
@ -0,0 +1,20 @@
|
|||
{
|
||||
"title": "Get Timezone",
|
||||
"description": "Form to select a timezone.",
|
||||
"type": "object",
|
||||
"required": [
|
||||
"timezone"
|
||||
],
|
||||
"properties": {
|
||||
"timezone": {
|
||||
"type": "string",
|
||||
"title": "Timezone",
|
||||
"enum": [
|
||||
"US/Eastern",
|
||||
"US/Pacific",
|
||||
"Europe/Berlin",
|
||||
"Australia/ACT"
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,59 @@
|
|||
from spiffworkflow_backend.services.process_instance_processor import ProcessInstanceProcessor
|
||||
from spiffworkflow_backend.services.process_instance_service import ProcessInstanceService
|
||||
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
|
||||
from flask.app import Flask
|
||||
from flask.testing import FlaskClient
|
||||
from spiffworkflow_backend.models.user import UserModel
|
||||
from spiffworkflow_backend.scripts.get_localtime import GetLocaltime
|
||||
import datetime
|
||||
import pytz
|
||||
|
||||
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
|
||||
|
||||
|
||||
class TestGetLocaltime(BaseTest):
|
||||
"""TestProcessAPi."""
|
||||
|
||||
def test_get_localtime_script_directly(self):
|
||||
current_time = datetime.datetime.now()
|
||||
timezone = "US/Pacific"
|
||||
result = GetLocaltime().run(task=None, environment_identifier='testing', datetime=current_time, timezone=timezone)
|
||||
assert result == current_time.astimezone(pytz.timezone(timezone))
|
||||
|
||||
def test_get_localtime_script_through_bpmn(
|
||||
self,
|
||||
app: Flask,
|
||||
client: FlaskClient,
|
||||
with_db_and_bpmn_file_cleanup: None,
|
||||
) -> None:
|
||||
"""Test_process_instance_run."""
|
||||
initiator_user = self.find_or_create_user("initiator_user")
|
||||
process_model = load_test_spec(
|
||||
process_model_id="get_localtime", bpmn_file_name="get_localtime.bpmn"
|
||||
)
|
||||
process_instance = self.create_process_instance_from_process_model(
|
||||
process_model=process_model, user=initiator_user
|
||||
)
|
||||
processor = ProcessInstanceProcessor(process_instance)
|
||||
|
||||
processor.do_engine_steps(save=True)
|
||||
active_task = process_instance.active_tasks[0]
|
||||
spiff_task = processor.__class__.get_task_by_bpmn_identifier(
|
||||
active_task.task_name, processor.bpmn_process_instance
|
||||
)
|
||||
|
||||
ProcessInstanceService.complete_form_task(
|
||||
processor, spiff_task, {"timezone": "US/Pacific"}, initiator_user
|
||||
)
|
||||
|
||||
active_task = process_instance.active_tasks[0]
|
||||
spiff_task = processor.__class__.get_task_by_bpmn_identifier(
|
||||
active_task.task_name, processor.bpmn_process_instance
|
||||
)
|
||||
|
||||
data = spiff_task.data
|
||||
some_time = data['some_time']
|
||||
localtime = data['localtime']
|
||||
timezone = data['timezone']
|
||||
|
||||
assert localtime == some_time.astimezone(pytz.timezone(timezone))
|
||||
Loading…
Reference in New Issue