Implement reverseOrder option for GetWorkflowExecutionHistory and PollForDecisionTask

This commit is contained in:
Jean-Baptiste Barth 2015-10-11 19:14:31 +02:00
parent c16da9da2d
commit aa4adbb76e
6 changed files with 40 additions and 15 deletions

View File

@ -16,10 +16,11 @@ class DecisionTask(object):
# but that shouldn't be a problem for tests # but that shouldn't be a problem for tests
self.scheduled_at = datetime.now() self.scheduled_at = datetime.now()
def to_full_dict(self): def to_full_dict(self, reverse_order=False):
events = self.workflow_execution.events(reverse_order=reverse_order)
hsh = { hsh = {
"events": [ "events": [
evt.to_dict() for evt in self.workflow_execution.events evt.to_dict() for evt in events
], ],
"taskToken": self.task_token, "taskToken": self.task_token,
"previousStartedEventId": self.previous_started_event_id, "previousStartedEventId": self.previous_started_event_id,

View File

@ -32,7 +32,7 @@ class WorkflowExecution(object):
"openChildWorkflowExecutions": 0, "openChildWorkflowExecutions": 0,
} }
# events # events
self.events = [] self._events = []
# tasks # tasks
self.decision_tasks = [] self.decision_tasks = []
self.activity_tasks = [] self.activity_tasks = []
@ -97,13 +97,19 @@ class WorkflowExecution(object):
hsh["openCounts"] = self.open_counts hsh["openCounts"] = self.open_counts
return hsh return hsh
def events(self, reverse_order=False):
if reverse_order:
return reversed(self._events)
else:
return self._events
def next_event_id(self): def next_event_id(self):
event_ids = [evt.event_id for evt in self.events] event_ids = [evt.event_id for evt in self._events]
return max(event_ids or [0]) return max(event_ids or [0])
def _add_event(self, *args, **kwargs): def _add_event(self, *args, **kwargs):
evt = HistoryEvent(self.next_event_id(), *args, **kwargs) evt = HistoryEvent(self.next_event_id(), *args, **kwargs)
self.events.append(evt) self._events.append(evt)
return evt return evt
def start(self): def start(self):

View File

@ -215,22 +215,24 @@ class SWFResponse(BaseResponse):
_workflow_execution = self._params["execution"] _workflow_execution = self._params["execution"]
run_id = _workflow_execution["runId"] run_id = _workflow_execution["runId"]
workflow_id = _workflow_execution["workflowId"] workflow_id = _workflow_execution["workflowId"]
# TODO: implement reverseOrder reverse_order = self._params.get("reverseOrder", None)
wfe = self.swf_backend.describe_workflow_execution(domain_name, run_id, workflow_id) wfe = self.swf_backend.describe_workflow_execution(domain_name, run_id, workflow_id)
events = wfe.events(reverse_order=reverse_order)
return json.dumps({ return json.dumps({
"events": [evt.to_dict() for evt in wfe.events] "events": [evt.to_dict() for evt in events]
}) })
def poll_for_decision_task(self): def poll_for_decision_task(self):
domain_name = self._params["domain"] domain_name = self._params["domain"]
task_list = self._params["taskList"]["name"] task_list = self._params["taskList"]["name"]
identity = self._params.get("identity") identity = self._params.get("identity")
# TODO: implement reverseOrder reverse_order = self._params.get("reverseOrder", None)
decision = self.swf_backend.poll_for_decision_task( decision = self.swf_backend.poll_for_decision_task(
domain_name, task_list, identity=identity domain_name, task_list, identity=identity
) )
if decision: if decision:
return json.dumps(decision.to_full_dict()) return json.dumps(
decision.to_full_dict(reverse_order=reverse_order)
)
else: else:
return json.dumps({"previousStartedEventId": 0, "startedEventId": 0}) return json.dumps({"previousStartedEventId": 0, "startedEventId": 0})

View File

@ -44,3 +44,10 @@ def test_poll_for_decision_task_when_none():
# this is the DecisionTask representation you get from the real SWF # this is the DecisionTask representation you get from the real SWF
# after waiting 60s when there's no decision to be taken # after waiting 60s when there's no decision to be taken
resp.should.equal({"previousStartedEventId": 0, "startedEventId": 0}) resp.should.equal({"previousStartedEventId": 0, "startedEventId": 0})
@mock_swf
def test_poll_for_decision_task_with_reverse_order():
conn = setup_workflow()
resp = conn.poll_for_decision_task("test-domain", "queue", reverse_order=True)
types = [evt["eventType"] for evt in resp["events"]]
types.should.equal(["DecisionTaskStarted", "DecisionTaskScheduled", "WorkflowExecutionStarted"])

View File

@ -197,8 +197,8 @@ def test_workflow_execution_start_decision_task():
wfe.start_decision_task(dt.task_token, identity="srv01") wfe.start_decision_task(dt.task_token, identity="srv01")
dt = wfe.decision_tasks[0] dt = wfe.decision_tasks[0]
dt.state.should.equal("STARTED") dt.state.should.equal("STARTED")
wfe.events[-1].event_type.should.equal("DecisionTaskStarted") wfe.events()[-1].event_type.should.equal("DecisionTaskStarted")
wfe.events[-1].identity.should.equal("srv01") wfe.events()[-1].identity.should.equal("srv01")
# HistoryEvent # HistoryEvent

View File

@ -79,10 +79,19 @@ def test_get_workflow_execution_history():
run_id = hsh["runId"] run_id = hsh["runId"]
resp = conn.get_workflow_execution_history("test-domain", run_id, "uid-abcd1234") resp = conn.get_workflow_execution_history("test-domain", run_id, "uid-abcd1234")
resp["events"].should.be.a("list") types = [evt["eventType"] for evt in resp["events"]]
evt = resp["events"][0] types.should.equal(["WorkflowExecutionStarted", "DecisionTaskScheduled"])
evt["eventType"].should.equal("WorkflowExecutionStarted")
@mock_swf
def test_get_workflow_execution_history_with_reverse_order():
conn = setup_swf_environment()
hsh = conn.start_workflow_execution("test-domain", "uid-abcd1234", "test-workflow", "v1.0")
run_id = hsh["runId"]
resp = conn.get_workflow_execution_history("test-domain", run_id, "uid-abcd1234",
reverse_order=True)
types = [evt["eventType"] for evt in resp["events"]]
types.should.equal(["DecisionTaskScheduled", "WorkflowExecutionStarted"])
@mock_swf @mock_swf
def test_get_workflow_execution_history_on_non_existent_workflow_execution(): def test_get_workflow_execution_history_on_non_existent_workflow_execution():