fix tests for waiting_event_can_be_skipped
This commit is contained in:
parent
440a4850f1
commit
2900a9a5f8
|
@ -10,6 +10,7 @@ from urllib.parse import unquote
|
|||
import sentry_sdk
|
||||
from flask import current_app
|
||||
from flask import g
|
||||
from SpiffWorkflow.bpmn.event import PendingBpmnEvent # type: ignore
|
||||
from SpiffWorkflow.bpmn.specs.control import BoundaryEventSplit # type: ignore
|
||||
from SpiffWorkflow.bpmn.specs.event_definitions.timer import TimerEventDefinition # type: ignore
|
||||
from SpiffWorkflow.task import Task as SpiffTask # type: ignore
|
||||
|
@ -144,9 +145,8 @@ class ProcessInstanceService:
|
|||
db.session.add(cycle)
|
||||
db.session.commit()
|
||||
|
||||
# TODO: test me
|
||||
@classmethod
|
||||
def waiting_event_can_be_skipped(cls, waiting_event: dict[str, Any], now_in_utc: datetime) -> bool:
|
||||
def waiting_event_can_be_skipped(cls, waiting_event: PendingBpmnEvent, now_in_utc: datetime) -> bool:
|
||||
#
|
||||
# over time this function can gain more knowledge of different event types,
|
||||
# for now we are just handling Duration Timer events.
|
||||
|
|
|
@ -2,6 +2,7 @@ from datetime import datetime
|
|||
from datetime import timezone
|
||||
|
||||
from flask.app import Flask
|
||||
from SpiffWorkflow.bpmn.event import PendingBpmnEvent # type: ignore
|
||||
from spiffworkflow_backend.models.process_instance_file_data import ProcessInstanceFileDataModel
|
||||
from spiffworkflow_backend.services.process_instance_service import ProcessInstanceService
|
||||
|
||||
|
@ -198,39 +199,47 @@ class TestProcessInstanceService(BaseTest):
|
|||
self._check_sample_file_data_model("File", 1, models[1])
|
||||
|
||||
def test_does_not_skip_events_it_does_not_know_about(self) -> None:
|
||||
name = None
|
||||
event_type = "Unknown"
|
||||
value = "2023-04-27T20:15:10.626656+00:00"
|
||||
pending_event = PendingBpmnEvent(name, event_type, value)
|
||||
assert not (
|
||||
ProcessInstanceService.waiting_event_can_be_skipped(
|
||||
{"event_type": "Unknown", "name": None, "value": "2023-04-27T20:15:10.626656+00:00"},
|
||||
pending_event,
|
||||
datetime.now(timezone.utc),
|
||||
)
|
||||
)
|
||||
|
||||
def test_does_skip_duration_timer_events_for_the_future(self) -> None:
|
||||
name = None
|
||||
event_type = "DurationTimerEventDefinition"
|
||||
value = "2023-04-27T20:15:10.626656+00:00"
|
||||
pending_event = PendingBpmnEvent(name, event_type, value)
|
||||
assert ProcessInstanceService.waiting_event_can_be_skipped(
|
||||
{"event_type": "DurationTimerEventDefinition", "name": None, "value": "2023-04-27T20:15:10.626656+00:00"},
|
||||
pending_event,
|
||||
datetime.fromisoformat("2023-04-26T20:15:10.626656+00:00"),
|
||||
)
|
||||
|
||||
def test_does_not_skip_duration_timer_events_for_the_past(self) -> None:
|
||||
name = None
|
||||
event_type = "DurationTimerEventDefinition"
|
||||
value = "2023-04-27T20:15:10.626656+00:00"
|
||||
pending_event = PendingBpmnEvent(name, event_type, value)
|
||||
assert not (
|
||||
ProcessInstanceService.waiting_event_can_be_skipped(
|
||||
{
|
||||
"event_type": "DurationTimerEventDefinition",
|
||||
"name": None,
|
||||
"value": "2023-04-27T20:15:10.626656+00:00",
|
||||
},
|
||||
pending_event,
|
||||
datetime.fromisoformat("2023-04-28T20:15:10.626656+00:00"),
|
||||
)
|
||||
)
|
||||
|
||||
def test_does_not_skip_duration_timer_events_for_now(self) -> None:
|
||||
name = None
|
||||
event_type = "DurationTimerEventDefinition"
|
||||
value = "2023-04-27T20:15:10.626656+00:00"
|
||||
pending_event = PendingBpmnEvent(name, event_type, value)
|
||||
assert not (
|
||||
ProcessInstanceService.waiting_event_can_be_skipped(
|
||||
{
|
||||
"event_type": "DurationTimerEventDefinition",
|
||||
"name": None,
|
||||
"value": "2023-04-27T20:15:10.626656+00:00",
|
||||
},
|
||||
pending_event,
|
||||
datetime.fromisoformat("2023-04-27T20:15:10.626656+00:00"),
|
||||
)
|
||||
)
|
||||
|
|
Loading…
Reference in New Issue