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
|
import boto.swf.layer2
from boto.swf.layer2 import ActivityType, WorkflowType, WorkflowExecution
from tests.unit import unittest
from mock import Mock, ANY
class TestTypes(unittest.TestCase):
def setUp(self):
boto.swf.layer2.Layer1 = Mock()
def test_workflow_type_register_defaults(self):
wf_type = WorkflowType(name='name', domain='test', version='1')
wf_type.register()
wf_type._swf.register_workflow_type.assert_called_with('test', 'name', '1',
default_execution_start_to_close_timeout=ANY,
default_task_start_to_close_timeout=ANY,
default_child_policy=ANY
)
def test_activity_type_register_defaults(self):
act_type = ActivityType(name='name', domain='test', version='1')
act_type.register()
act_type._swf.register_activity_type.assert_called_with('test', 'name', '1',
default_task_heartbeat_timeout=ANY,
default_task_schedule_to_close_timeout=ANY,
default_task_schedule_to_start_timeout=ANY,
default_task_start_to_close_timeout=ANY
)
def test_workflow_type_start_execution(self):
wf_type = WorkflowType(name='name', domain='test', version='1')
run_id = '122aJcg6ic7MRAkjDRzLBsqU/R49qt5D0LPHycT/6ArN4='
wf_type._swf.start_workflow_execution.return_value = {'runId': run_id}
execution = wf_type.start(task_list='hello_world')
self.assertIsInstance(execution, WorkflowExecution)
self.assertEquals(wf_type.name, execution.name)
self.assertEquals(wf_type.version, execution.version)
self.assertEquals(run_id, execution.runId)
if __name__ == '__main__':
unittest.main()
|
