Add SWF endpoint: StartWorkflowExecution

This commit is contained in:
Jean-Baptiste Barth 2015-10-02 05:03:10 +02:00
parent fbcdd5f2bd
commit 92cf64c2ad
5 changed files with 177 additions and 6 deletions

View File

@ -56,3 +56,10 @@ class SWFTypeDeprecatedFault(SWFClientError):
"{}=[name={}, version={}]".format(_type.__class__.__name__, _type.name, _type.version),
"com.amazonaws.swf.base.model#TypeDeprecatedFault")
class SWFWorkflowExecutionAlreadyStartedFault(JSONResponseError):
def __init__(self):
super(SWFWorkflowExecutionAlreadyStartedFault, self).__init__(
400, "Bad Request",
body={"__type": "com.amazonaws.swf.base.model#WorkflowExecutionAlreadyStartedFault"}
)

View File

@ -1,5 +1,6 @@
from __future__ import unicode_literals
from collections import defaultdict
import uuid
import boto.swf
@ -13,6 +14,7 @@ from .exceptions import (
SWFSerializationException,
SWFTypeAlreadyExistsFault,
SWFTypeDeprecatedFault,
SWFWorkflowExecutionAlreadyStartedFault,
)
@ -26,6 +28,12 @@ class Domain(object):
"activity": defaultdict(dict),
"workflow": defaultdict(dict),
}
# Workflow executions have an id, which unicity is guaranteed
# at domain level (not super clear in the docs, but I checked
# that against SWF API) ; hence the storage method as a dict
# of "workflow_id (client determined)" => WorkflowExecution()
# here.
self.workflow_executions = {}
def __repr__(self):
return "Domain(name: %(name)s, status: %(status)s)" % self.__dict__
@ -62,6 +70,11 @@ class Domain(object):
_all.append(_type)
return _all
def add_workflow_execution(self, workflow_execution_id, workflow_execution):
if self.workflow_executions.get(workflow_execution_id):
raise SWFWorkflowExecutionAlreadyStartedFault()
self.workflow_executions[workflow_execution_id] = workflow_execution
class GenericType(object):
def __init__(self, name, version, **kwargs):
@ -120,6 +133,7 @@ class GenericType(object):
hsh["configuration"][key] = getattr(self, attr)
return hsh
class ActivityType(GenericType):
@property
def _configuration_keys(self):
@ -149,6 +163,17 @@ class WorkflowType(GenericType):
return "workflow"
class WorkflowExecution(object):
def __init__(self, workflow_type, **kwargs):
self.workflow_type = workflow_type
self.run_id = uuid.uuid4().hex
for key, value in kwargs.iteritems():
self.__setattr__(key, value)
def __repr__(self):
return "WorkflowExecution(run_id: {})".format(self.run_id)
class SWFBackend(BaseBackend):
def __init__(self, region_name):
self.region_name = region_name
@ -168,10 +193,25 @@ class SWFBackend(BaseBackend):
return matching[0]
return None
def _check_none_or_string(self, parameter):
if parameter is not None:
self._check_string(parameter)
def _check_string(self, parameter):
if not isinstance(parameter, basestring):
raise SWFSerializationException(parameter)
def _check_none_or_list_of_strings(self, parameter):
if parameter is not None:
self._check_list_of_strings(parameter)
def _check_list_of_strings(self, parameter):
if not isinstance(parameter, list):
raise SWFSerializationException(parameter)
for i in parameter:
if not isinstance(i, basestring):
raise SWFSerializationException(parameter)
def list_domains(self, status, reverse_order=None):
self._check_string(status)
domains = [domain for domain in self.domains
@ -185,8 +225,7 @@ class SWFBackend(BaseBackend):
description=None):
self._check_string(name)
self._check_string(workflow_execution_retention_period_in_days)
if description:
self._check_string(description)
self._check_none_or_string(description)
if self._get_domain(name, ignore_empty=True):
raise SWFDomainAlreadyExistsFault(name)
domain = Domain(name, workflow_execution_retention_period_in_days,
@ -219,10 +258,7 @@ class SWFBackend(BaseBackend):
self._check_string(name)
self._check_string(version)
for _, value in kwargs.iteritems():
if value == (None,):
print _
if value is not None:
self._check_string(value)
self._check_none_or_string(value)
domain = self._get_domain(domain_name)
_type = domain.get_type(kind, name, version, ignore_empty=True)
if _type:
@ -248,6 +284,29 @@ class SWFBackend(BaseBackend):
domain = self._get_domain(domain_name)
return domain.get_type(kind, name, version)
# TODO: find what triggers a "DefaultUndefinedFault" and implement it
# (didn't found in boto source code, nor in the docs, nor on a Google search)
# (will try to reach support)
def start_workflow_execution(self, domain_name, workflow_execution_id,
workflow_name, workflow_version,
tag_list=None, **kwargs):
self._check_string(domain_name)
self._check_string(workflow_execution_id)
self._check_string(workflow_name)
self._check_string(workflow_version)
self._check_none_or_list_of_strings(tag_list)
for _, value in kwargs.iteritems():
self._check_none_or_string(value)
domain = self._get_domain(domain_name)
wf_type = domain.get_type("workflow", workflow_name, workflow_version)
if wf_type.status == "DEPRECATED":
raise SWFTypeDeprecatedFault(wf_type)
wfe = WorkflowExecution(wf_type, tag_list=tag_list, **kwargs)
domain.add_workflow_execution(workflow_execution_id, wfe)
return wfe
swf_backends = {}
for region in boto.swf.regions():

View File

@ -175,3 +175,32 @@ class SWFResponse(BaseResponse):
def describe_workflow_type(self):
return self._describe_type("workflow")
def start_workflow_execution(self):
domain = self._params["domain"]
workflow_id = self._params["workflowId"]
_workflow_type = self._params["workflowType"]
workflow_name = _workflow_type["name"]
workflow_version = _workflow_type["version"]
_default_task_list = self._params.get("defaultTaskList")
if _default_task_list:
task_list = _default_task_list.get("name")
else:
task_list = None
child_policy = self._params.get("childPolicy")
execution_start_to_close_timeout = self._params.get("executionStartToCloseTimeout")
input_ = self._params.get("input")
tag_list = self._params.get("tagList")
task_start_to_close_timeout = self._params.get("taskStartToCloseTimeout")
wfe = self.swf_backend.start_workflow_execution(
domain, workflow_id, workflow_name, workflow_version,
task_list=task_list, child_policy=child_policy,
execution_start_to_close_timeout=execution_start_to_close_timeout,
input=input_, tag_list=tag_list,
task_start_to_close_timeout=task_start_to_close_timeout
)
return json.dumps({
"runId": wfe.run_id
})

View File

@ -3,6 +3,7 @@ from sure import expect
from moto.swf.models import (
Domain,
GenericType,
WorkflowExecution,
)
@ -65,3 +66,19 @@ def test_type_full_dict_representation():
def test_type_string_representation():
_type = FooType("test-foo", "v1.0")
str(_type).should.equal("FooType(name: test-foo, version: v1.0, status: REGISTERED)")
# WorkflowExecution
def test_workflow_execution_creation():
wfe = WorkflowExecution("workflow_type_whatever", child_policy="TERMINATE")
wfe.workflow_type.should.equal("workflow_type_whatever")
wfe.child_policy.should.equal("TERMINATE")
def test_workflow_execution_string_representation():
wfe = WorkflowExecution("workflow_type_whatever", child_policy="TERMINATE")
str(wfe).should.match(r"^WorkflowExecution\(run_id: .*\)")
def test_workflow_execution_generates_a_random_run_id():
wfe1 = WorkflowExecution("workflow_type_whatever")
wfe2 = WorkflowExecution("workflow_type_whatever")
wfe1.run_id.should_not.equal(wfe2.run_id)

View File

@ -0,0 +1,59 @@
import boto
from nose.tools import assert_raises
from sure import expect
from moto import mock_swf
from moto.swf.exceptions import (
SWFWorkflowExecutionAlreadyStartedFault,
SWFTypeDeprecatedFault,
)
# Utils
@mock_swf
def setup_swf_environment():
conn = boto.connect_swf("the_key", "the_secret")
conn.register_domain("test-domain", "60", description="A test domain")
conn.register_workflow_type("test-domain", "test-workflow", "v1.0")
conn.register_activity_type("test-domain", "test-activity", "v1.1")
return conn
# StartWorkflowExecution endpoint
@mock_swf
def test_start_workflow_execution():
conn = setup_swf_environment()
wf = conn.start_workflow_execution("test-domain", "uid-abcd1234", "test-workflow", "v1.0")
wf.should.contain("runId")
@mock_swf
def test_start_already_started_workflow_execution():
conn = setup_swf_environment()
conn.start_workflow_execution("test-domain", "uid-abcd1234", "test-workflow", "v1.0")
with assert_raises(SWFWorkflowExecutionAlreadyStartedFault) as err:
conn.start_workflow_execution("test-domain", "uid-abcd1234", "test-workflow", "v1.0")
ex = err.exception
ex.status.should.equal(400)
ex.error_code.should.equal("WorkflowExecutionAlreadyStartedFault")
ex.body.should.equal({
"__type": "com.amazonaws.swf.base.model#WorkflowExecutionAlreadyStartedFault",
})
@mock_swf
def test_start_workflow_execution_on_deprecated_type():
conn = setup_swf_environment()
conn.deprecate_workflow_type("test-domain", "test-workflow", "v1.0")
with assert_raises(SWFTypeDeprecatedFault) as err:
conn.start_workflow_execution("test-domain", "uid-abcd1234", "test-workflow", "v1.0")
ex = err.exception
ex.status.should.equal(400)
ex.error_code.should.equal("TypeDeprecatedFault")
ex.body.should.equal({
"__type": "com.amazonaws.swf.base.model#TypeDeprecatedFault",
"message": "WorkflowType=[name=test-workflow, version=v1.0]"
})