catalin/moto
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Add SWF endpoint TerminateWorkflowExecution

Browse source
This commit is contained in:
Jean-Baptiste Barth 2015-11-01 21:55:07 +01:00
commit 96d6bb056b
10 changed files with 212 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

20
moto/swf/models/__init__.py
View file

@ -160,7 +160,7 @@ class SWFBackend(BaseBackend):
self._check_string(run_id)
self._check_string(workflow_id)
domain = self._get_domain(domain_name)
return domain.get_workflow_execution(run_id, workflow_id)
return domain.get_workflow_execution(workflow_id, run_id=run_id)
def poll_for_decision_task(self, domain_name, task_list, identity=None):
self._check_string(domain_name)
@ -198,7 +198,7 @@ class SWFBackend(BaseBackend):
self._check_string(task_list)
domain = self._get_domain(domain_name)
count = 0
for _, wfe in domain.workflow_executions.iteritems():
for wfe in domain.workflow_executions:
if wfe.task_list == task_list:
count += wfe.open_counts["openDecisionTasks"]
return count
@ -211,7 +211,7 @@ class SWFBackend(BaseBackend):
# let's find decision task
decision_task = None
for domain in self.domains:
for _, wfe in domain.workflow_executions.iteritems():
for wfe in domain.workflow_executions:
for dt in wfe.decision_tasks:
if dt.task_token == task_token:
decision_task = dt
@ -301,7 +301,7 @@ class SWFBackend(BaseBackend):
def _find_activity_task_from_token(self, task_token):
activity_task = None
for domain in self.domains:
for _, wfe in domain.workflow_executions.iteritems():
for wfe in domain.workflow_executions:
for task in wfe.activity_tasks:
if task.task_token == task_token:
activity_task = task
@ -352,6 +352,18 @@ class SWFBackend(BaseBackend):
wfe = activity_task.workflow_execution
wfe.fail_activity_task(activity_task.task_token, reason=reason, details=details)
def terminate_workflow_execution(self, domain_name, workflow_id, child_policy=None,
details=None, reason=None, run_id=None):
self._check_string(domain_name)
self._check_string(workflow_id)
self._check_none_or_string(child_policy)
self._check_none_or_string(details)
self._check_none_or_string(reason)
self._check_none_or_string(run_id)
domain = self._get_domain(domain_name)
wfe = domain.get_workflow_execution(workflow_id, run_id=run_id, raise_if_closed=True)
wfe.terminate(child_policy=child_policy, details=details, reason=reason)
swf_backends = {}
for region in boto.swf.regions():

34
moto/swf/models/domain.py
View file

@ -22,7 +22,7 @@ class Domain(object):
# that against SWF API) ; hence the storage method as a dict
# of "workflow_id (client determined)" => WorkflowExecution()
# here.
self.workflow_executions = {}
self.workflow_executions = []
self.activity_task_lists = {}
self.decision_task_lists = {}
@ -71,18 +71,32 @@ class Domain(object):
def add_workflow_execution(self, workflow_execution):
_id = workflow_execution.workflow_id
if self.workflow_executions.get(_id):
# TODO: handle this better: this should raise ONLY if there's an OPEN wfe with this ID
if any(wfe.workflow_id == _id for wfe in self.workflow_executions):
raise SWFWorkflowExecutionAlreadyStartedFault()
self.workflow_executions[_id] = workflow_execution
self.workflow_executions.append(workflow_execution)
def get_workflow_execution(self, run_id, workflow_id):
wfe = self.workflow_executions.get(workflow_id)
if not wfe or wfe.run_id != run_id:
raise SWFUnknownResourceFault(
"execution",
"WorkflowExecution=[workflowId={}, runId={}]".format(
workflow_id, run_id
def get_workflow_execution(self, workflow_id, run_id=None, raise_if_closed=False):
if run_id:
_all = [w for w in self.workflow_executions
if w.workflow_id == workflow_id and w.run_id == run_id]
else:
_all = [w for w in self.workflow_executions
if w.workflow_id == workflow_id and w.execution_status == "OPEN"]
wfe = _all[0] if _all else None
if raise_if_closed and wfe and wfe.execution_status == "CLOSED":
wfe = None
if run_id:
if not wfe or wfe.run_id != run_id:
raise SWFUnknownResourceFault(
"execution",
"WorkflowExecution=[workflowId={}, runId={}]".format(
workflow_id, run_id
)
)
elif not wfe:
raise SWFUnknownResourceFault(
"execution, workflowId = {}".format(workflow_id)
)
return wfe

11
moto/swf/models/history_event.py
View file

@ -132,6 +132,17 @@ class HistoryEvent(object):
if hasattr(self, "details") and self.details is not None:
hsh["details"] = self.details
return hsh
elif self.event_type == "WorkflowExecutionTerminated":
hsh = {
"childPolicy": self.child_policy,
}
if self.cause:
hsh["cause"] = self.cause
if self.details:
hsh["details"] = self.details
if self.reason:
hsh["reason"] = self.reason
return hsh
else:
raise NotImplementedError(
"HistoryEvent does not implement attributes for type '{}'".format(self.event_type)

19
moto/swf/models/workflow_execution.py
View file

@ -50,6 +50,7 @@ class WorkflowExecution(object):
# TODO: check valid values among:
# COMPLETED | FAILED | CANCELED | TERMINATED | CONTINUED_AS_NEW | TIMED_OUT
# TODO: implement them all
self.close_cause = None
self.close_status = None
self.close_timestamp = None
self.execution_status = "OPEN"
@ -467,3 +468,21 @@ class WorkflowExecution(object):
self.open_counts["openActivityTasks"] -= 1
# TODO: ensure we don't schedule multiple decisions at the same time!
self.schedule_decision_task()
def terminate(self, child_policy=None, details=None, reason=None):
# TODO: handle child policy for child workflows here
# TODO: handle cause="CHILD_POLICY_APPLIED"
# Until this, we set cause manually to "OPERATOR_INITIATED"
cause = "OPERATOR_INITIATED"
if not child_policy:
child_policy = self.child_policy
self._add_event(
"WorkflowExecutionTerminated",
cause=cause,
child_policy=child_policy,
details=details,
reason=reason,
)
self.execution_status = "CLOSED"
self.close_status = "TERMINATED"
self.close_cause = "OPERATOR_INITIATED"