Merge pull request #19 from botify-labs/fix/schedule-failed

Fix/schedule failed
botify-labs · Aug 7, 2014 · 70b72cf · 70b72cf
2 parents 39e7594 + 50a30c9
commit 70b72cf
Show file tree

Hide file tree

Showing 3 changed files with 49 additions and 5 deletions.
diff --git a/simpleflow/history.py b/simpleflow/history.py
@@ -1,8 +1,13 @@
+import collections
+
+
 class History(object):
     def __init__(self, history):
         self._history = history
-        self._activities = {}
-        self._child_workflows = {}
+        self._activities = collections.defaultdict(
+            lambda: {'type': 'activity'})
+        self._child_workflows = collections.defaultdict(
+            lambda: {'type': 'child_workflow'})
 
     @property
     def events(self):

diff --git a/simpleflow/swf/executor.py b/simpleflow/swf/executor.py
@@ -5,6 +5,7 @@
 
 import swf.models
 import swf.models.decision
+import swf.exceptions
 
 from simpleflow import (
     executor,

diff --git a/tests/test_dataflow.py b/tests/test_dataflow.py
@@ -1,6 +1,9 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 
+import functools
+import mock
+
 import boto.swf
 
 
@@ -955,11 +958,9 @@ def test_activity_not_found_schedule_failed():
     history = builder.History(workflow)
     decision_id = history.last_id
     (history
-        .add_activity_task(
-            raise_on_failure,
+        .add_activity_task_schedule_failed(
             activity_id='activity-tests.test_dataflow.increment-1',
             decision_id=decision_id,
-            last_state='schedule_failed',
             activity_type={
                 'name': increment.name,
                 'version': increment.version
@@ -968,3 +969,40 @@ def test_activity_not_found_schedule_failed():
 
     decisions, _ = executor.replay(history)
     check_task_scheduled_decision(decisions[0], increment)
+
+
+def raise_already_exists(activity):
+    @functools.wraps(raise_already_exists)
+    def wrapped(*args):
+        raise swf.exceptions.AlreadyExistsError(
+            '<ActivityType domain={} name={} version={} status=REGISTERED> '
+            'already exists'.format(
+                DOMAIN.name,
+                activity.name,
+                activity.version))
+
+    return wrapped
+
+
+def test_activity_not_found_schedule_failed_already_exists():
+    workflow = TestDefinition
+    executor = Executor(DOMAIN, workflow)
+
+    history = builder.History(workflow)
+    decision_id = history.last_id
+    (history
+        .add_activity_task_schedule_failed(
+            activity_id='activity-tests.test_dataflow.increment-1',
+            decision_id=decision_id,
+            activity_type={
+                'name': increment.name,
+                'version': increment.version
+            },
+            cause='ACTIVITY_TYPE_DOES_NOT_EXIST'))
+
+    with mock.patch(
+            'swf.models.ActivityType.save',
+            raise_already_exists(increment)):
+        decisions, _ = executor.replay(history)
+
+    check_task_scheduled_decision(decisions[0], increment)