serverless-workflow-manager/test_workflow.py at master · jeyabalajis/serverless-workflow-manager · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
import json
from unittest import TestCase

from domain.stage import Stage
from domain.workflow import Workflow


class TestWorkflow(TestCase):

    @classmethod
    def get_sample_workflow_instance(cls):
        with open("samples/sample_workflow_instance.json") as workflow_instance_file:
            return json.load(workflow_instance_file)

    def test_all_dependencies_completed_false(self):
        workflow_object = Workflow.from_json(workflow_json=self.get_sample_workflow_instance())

        stage = workflow_object.get_stage_by_name(stage_name="PREPARE")
        task = workflow_object.get_task_by_name(stage=stage, task_name="confirm_delivery")
        all_deps_completed = workflow_object.all_dependencies_completed_for_a_task(stage=stage, task=task)

        assert all_deps_completed is False

    def test_all_dependencies_completed_true(self):
        workflow_object = Workflow.from_json(workflow_json=self.get_sample_workflow_instance())

        stage = workflow_object.get_stage_by_name(stage_name="ORDER")
        task = workflow_object.get_task_by_name(stage=stage, task_name="confirm_order")
        all_deps_completed = workflow_object.all_dependencies_completed_for_a_task(stage=stage, task=task)

        assert all_deps_completed is True

    def test_get_active_stage(self):
        workflow_object = Workflow.from_json(workflow_json=self.get_sample_workflow_instance())
        active_stage = workflow_object.get_active_stage()
        assert active_stage.status == Stage.ACTIVE_STATUS

    def test_mark_stage_as_completed(self):
        workflow_object = Workflow.from_json(workflow_json=self.get_sample_workflow_instance())
        current_active_stage = workflow_object.get_active_stage()
        workflow_object.mark_stage_as_completed(stage=current_active_stage)
        next_active_stage = workflow_object.get_active_stage()

        print("current: {} next: {}".format(current_active_stage.stage_name, next_active_stage.stage_name))
        assert next_active_stage.stage_order == current_active_stage.stage_order + 1
        assert (
                current_active_stage.status == Stage.COMPLETED_STATUS
                and next_active_stage.status == Stage.ACTIVE_STATUS
        )

        workflow_object.mark_stage_as_completed(stage=next_active_stage)

        next_next_active_stage = workflow_object.get_active_stage()
        assert next_next_active_stage is None

    def test_find_and_schedule_tasks(self):
        workflow_object = Workflow.from_json(workflow_json=self.get_sample_workflow_instance())

        workflow_object.find_and_schedule_tasks()
        active_stage = workflow_object.get_active_stage()
        pending_tasks = active_stage.get_pending_tasks()
        assert pending_tasks is not None and len(pending_tasks) == 1

        scheduled_tasks = active_stage.get_scheduled_tasks()
        for task in scheduled_tasks:
            workflow_object.mark_task_as_completed(stage=active_stage, task=task)

        workflow_object.find_and_schedule_tasks()
        active_stage = workflow_object.get_active_stage()
        pending_tasks = active_stage.get_pending_tasks()
        scheduled_tasks = active_stage.get_scheduled_tasks()
        assert len(pending_tasks) == 0 and len(scheduled_tasks) == 1