Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Move setup_workflow() test function in test utils

Browse Source
This commit is contained in:
Jean-Baptiste Barth 2015-10-26 18:05:45 +01:00
parent be71909a8c
commit a0e484fa6d
2 changed files with 22 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
tests/test_swf/responses/test_decision_tasks.py
View File

@ -9,24 +9,7 @@ from moto.swf.exceptions import (
SWFDecisionValidationException,
)
from ..utils import mock_basic_workflow_type
@mock_swf
def setup_workflow():
conn = boto.connect_swf("the_key", "the_secret")
conn.register_domain("test-domain", "60", description="A test domain")
conn = mock_basic_workflow_type("test-domain", conn)
conn.register_activity_type(
"test-domain", "test-activity", "v1.1",
default_task_heartbeat_timeout="600",
default_task_schedule_to_close_timeout="600",
default_task_schedule_to_start_timeout="600",
default_task_start_to_close_timeout="600",
)
wfe = conn.start_workflow_execution("test-domain", "uid-abcd1234", "test-workflow", "v1.0")
conn.run_id = wfe["runId"]
return conn
from ..utils import setup_workflow
# PollForDecisionTask endpoint

21
tests/test_swf/utils.py
View File

@ -1,3 +1,6 @@
import boto
from moto import mock_swf
from moto.swf.models import (
ActivityType,
Domain,
@ -38,3 +41,21 @@ def make_workflow_execution(**kwargs):
domain.add_type(ActivityType("test-activity", "v1.1"))
wft = get_basic_workflow_type()
return WorkflowExecution(domain, wft, "ab1234", **kwargs)
# Setup a complete example workflow and return the connection object
@mock_swf
def setup_workflow():
conn = boto.connect_swf("the_key", "the_secret")
conn.register_domain("test-domain", "60", description="A test domain")
conn = mock_basic_workflow_type("test-domain", conn)
conn.register_activity_type(
"test-domain", "test-activity", "v1.1",
default_task_heartbeat_timeout="600",
default_task_schedule_to_close_timeout="600",
default_task_schedule_to_start_timeout="600",
default_task_start_to_close_timeout="600",
)
wfe = conn.start_workflow_execution("test-domain", "uid-abcd1234", "test-workflow", "v1.0")
conn.run_id = wfe["runId"]
return conn