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
|
from tests.unit import unittest
import boto.swf.layer1_decisions
class TestDecisions(unittest.TestCase):
def setUp(self):
self.decisions = boto.swf.layer1_decisions.Layer1Decisions()
def assert_data(self, *data):
self.assertEquals(self.decisions._data, list(data))
def test_continue_as_new_workflow_execution(self):
self.decisions.continue_as_new_workflow_execution(
child_policy='TERMINATE',
execution_start_to_close_timeout='10',
input='input',
tag_list=['t1', 't2'],
task_list='tasklist',
start_to_close_timeout='20',
workflow_type_version='v2'
)
self.assert_data({
'decisionType': 'ContinueAsNewWorkflowExecution',
'continueAsNewWorkflowExecutionDecisionAttributes': {
'childPolicy': 'TERMINATE',
'executionStartToCloseTimeout': '10',
'input': 'input',
'tagList': ['t1', 't2'],
'taskList': {'name': 'tasklist'},
'taskStartToCloseTimeout': '20',
'workflowTypeVersion': 'v2',
}
})
|
