diff --git a/test/workflows/__init__.py b/test/workflows/__init__.py index 320af9738138d715891bb6e1c575be41ea7df04f..19707ff0ff1b4e53e3309f9c3509c8e24001a9fc 100644 --- a/test/workflows/__init__.py +++ b/test/workflows/__init__.py @@ -2,7 +2,7 @@ import difflib import pprint from copy import deepcopy from itertools import chain, repeat -from typing import Callable, Tuple, cast +from typing import Callable, cast from uuid import uuid4 import structlog @@ -143,7 +143,7 @@ class WorkflowInstanceForTests(LazyWorkflowInstance): return f"WorkflowInstanceForTests('{self.workflow}','{self.name}')" -def _store_step(step_log: list[Tuple[Step, WFProcess]]) -> Callable[[ProcessStat, Step, WFProcess], WFProcess]: +def _store_step(step_log: list[tuple[Step, WFProcess]]) -> Callable[[ProcessStat, Step, WFProcess], WFProcess]: def __store_step(pstat: ProcessStat, step: Step, state: WFProcess) -> WFProcess: try: state = state.map(lambda s: json_loads(json_dumps(s))) @@ -164,14 +164,14 @@ def _sanitize_input(input_data: State | list[State]) -> list[State]: return cast(list[State], json_loads(json_dumps(input_data))) -def run_workflow(workflow_key: str, input_data: State | list[State]) -> Tuple[WFProcess, ProcessStat, list]: +def run_workflow(workflow_key: str, input_data: State | list[State]) -> tuple[WFProcess, ProcessStat, list]: # ATTENTION!! This code needs to be as similar as possible to `server.services.processes.start_process` # The main differences are: we use a different step log function and we don't run in # a sepperate thread user_data = _sanitize_input(input_data) user = "john.doe" - step_log: list[Tuple[Step, WFProcess]] = [] + step_log: list[tuple[Step, WFProcess]] = [] process_id = uuid4() workflow = get_workflow(workflow_key) @@ -201,8 +201,8 @@ def run_workflow(workflow_key: str, input_data: State | list[State]) -> Tuple[WF def resume_workflow( - process: ProcessStat, step_log: list[Tuple[Step, WFProcess]], input_data: State -) -> Tuple[WFProcess, list]: + process: ProcessStat, step_log: list[tuple[Step, WFProcess]], input_data: State +) -> tuple[WFProcess, list]: # ATTENTION!! This code needs to be as similar as possible to `server.services.processes.resume_process` # The main differences are: we use a different step log function, and we don't run in a separate thread user_data = _sanitize_input(input_data) @@ -229,7 +229,7 @@ def resume_workflow( def run_form_generator( form_generator: FormGenerator, extra_inputs: list[State] | None = None -) -> Tuple[list[dict], State]: +) -> tuple[list[dict], State]: """Run a form generator to get the resulting forms and result. Warning! This does not run the actual pydantic validation on purpose. However, you should @@ -245,7 +245,7 @@ def run_form_generator( Returns: ------- - Tuple[list[dict], State]: A list of generated forms and the result state for the whole generator. + tuple[list[dict], State]: A list of generated forms and the result state for the whole generator. Example: -------