Repository: botify-labs/simpleflow
Issue: TimeoutError has not attritube 'reason'

`swf.executor` [maps a timeout event to](https://github.com/botify-labs/simpleflow/blob/master/simpleflow/swf/executor.py#L110) a [`TimeoutError`](https://github.com/botify-labs/simpleflow/blob/master/simpleflow/exceptions.py#L46) object that does not have a `reason` attribute. If the task has the flag `raises_on_failure` enabled, `swf.Executor.resume()` will [wrap the `TimeoutError` in a `TaskException`](https://github.com/botify-labs/simpleflow/blob/master/simpleflow/swf/executor.py#L179). It then raises in `swf.Executor.replay()` that logs a message. It fails to [make the message](https://github.com/botify-labs/simpleflow/blob/master/simpleflow/swf/executor.py#L264) from the `reason` attribute.


Changed files:
setup.py, simpleflow/exceptions.py, tests/test_dataflow.py, simpleflow/swf/executor.py, simpleflow/task.py

diff --git a/setup.py b/setup.py
index 9443dcc9..9c80cb5c 100755
--- a/setup.py
+++ b/setup.py
@@ -81,7 +81,7 @@ setup(
     package_dir={'simpleflow': 'simpleflow'},
     include_package_data=True,
     install_requires=[
-        'simple-workflow>=0.1.39',
+        'simple-workflow>=0.1.40',
     ],
     license=read("LICENSE"),
     zip_safe=False,
diff --git a/simpleflow/exceptions.py b/simpleflow/exceptions.py
index 5c07e323..1ecd497b 100644
--- a/simpleflow/exceptions.py
+++ b/simpleflow/exceptions.py
@@ -7,17 +7,19 @@ class TaskException(Exception):
     Wrap an exception raised by a task.
 
     """
-    def __init__(self, exception):
+    def __init__(self, task, exception):
         """
         :param exception: raised by a task.
         :type  exception: TaskFailed.
 
         """
+        self.task = task
         self.exception = exception
 
     def __repr__(self):
-        return '{}(exception={})'.format(
+        return '{}(task={} exception={})'.format(
             self.__class__.__name__,
+            self.task,
             self.exception)
 
 
@@ -37,7 +39,7 @@ class TaskFailed(Exception):
         self.details = None
 
     def __repr__(self):
-        return '{}(reason="{}", details="{}")'.format(
+        return '{}(reason="{}")'.format(
             self.__class__.__name__,
             self.reason,
             self.details)
@@ -49,4 +51,6 @@ class TimeoutError(Exception):
         self.timeout_value = timeout_value
 
     def __repr__(self):
-        return 'TimeoutError({})'.format(self.timeout_type)
+        return '{}({})'.format(
+            self.__class__.__name__,
+            self.timeout_type)
diff --git a/simpleflow/swf/executor.py b/simpleflow/swf/executor.py
index 2510b05d..f66d68d9 100644
--- a/simpleflow/swf/executor.py
+++ b/simpleflow/swf/executor.py
@@ -177,7 +177,7 @@ class Executor(executor.Executor):
                     return future
                 elif event.get('retry', 0) == task.activity.retry:
                     if task.activity.raises_on_failure:
-                        raise exceptions.TaskException(future.exception)
+                        raise exceptions.TaskException(task, future.exception)
                     return future
                 # Otherwise retry the task by scheduling it again.
             elif event['type'] == 'child_workflow':
@@ -261,11 +261,12 @@ class Executor(executor.Executor):
         except exceptions.ExecutionBlocked:
             return self._decisions, {}
         except exceptions.TaskException, err:
-            reason = 'Workflow execution error: "{}"'.format(
-                err.exception.reason)
+            reason = 'Workflow execution error in task {}: "{}"'.format(
+                err.task.name,
+                getattr(err.exception, 'reason', repr(err.exception)))
             logger.exception(reason)
 
-            details = err.exception.details
+            details = getattr(err.exception, 'details', None)
             self.on_failure(reason, details)
 
             decision = swf.models.decision.WorkflowExecutionDecision()
diff --git a/simpleflow/task.py b/simpleflow/task.py
index 8b4cee90..fbf7a184 100644
--- a/simpleflow/task.py
+++ b/simpleflow/task.py
@@ -39,6 +39,14 @@ class ActivityTask(Task):
     def name(self):
         return 'activity-{}'.format(self.activity.name)
 
+    def __repr__(self):
+        return '{}(activity={}, args={}, kwargs={}, id={})'.format(
+            self.__class__.__name__,
+            self.activity,
+            self.args,
+            self.kwargs,
+            self.id)
+
 
 class WorkflowTask(Task):
     def __init__(self, workflow, *args, **kwargs):
@@ -51,6 +59,14 @@ class WorkflowTask(Task):
     def name(self):
         return 'workflow-{}'.format(self.workflow.name)
 
+    def __repr__(self):
+        return '{}(workflow={}, args={}, kwargs={}, id={})'.format(
+            self.__class__.__name__,
+            self.activity,
+            self.args,
+            self.kwargs,
+            self.id)
+
 
 class Registry(object):
     def __init__(self):
diff --git a/tests/test_dataflow.py b/tests/test_dataflow.py
index 3a13a86d..eb120ab0 100644
--- a/tests/test_dataflow.py
+++ b/tests/test_dataflow.py
@@ -735,7 +735,9 @@ def test_workflow_activity_raises_on_failure():
 
     workflow_failed = swf.models.decision.WorkflowExecutionDecision()
     workflow_failed.fail(
-        reason='Workflow execution error: "error"')
+        reason='Workflow execution error in task '
+               'activity-tests.test_dataflow.raise_on_failure: '
+               '"error"')
 
     assert decisions[0] == workflow_failed
 
@@ -821,3 +823,66 @@ def test_multiple_scheduled_activities():
 
     decisions, _ = executor.replay(history)
     check_task_scheduled_decision(decisions[0], double)
+
+
+def test_activity_task_timeout():
+    workflow = TestDefinition
+    executor = Executor(DOMAIN, workflow)
+
+    history = builder.History(workflow)
+    decision_id = history.last_id
+    (history
+        .add_activity_task(
+            increment,
+            activity_id='activity-tests.test_dataflow.increment-1',
+            decision_id=decision_id,
+            last_state='timed_out',
+            timeout_type='START_TO_CLOSE'))
+
+    decisions, _ = executor.replay(history)
+    # The task timed out and there is no retry.
+    assert len(decisions) == 1
+    check_task_scheduled_decision(decisions[0], double)
+
+
+def test_activity_task_timeout_retry():
+    workflow = TestDefinitionRetryActivity
+    executor = Executor(DOMAIN, workflow)
+
+    history = builder.History(workflow)
+    decision_id = history.last_id
+    (history
+        .add_activity_task(
+            increment_retry,
+            activity_id='activity-tests.test_dataflow.increment_retry-1',
+            decision_id=decision_id,
+            last_state='timed_out',
+            timeout_type='START_TO_CLOSE'))
+
+    decisions, _ = executor.replay(history)
+    assert len(decisions) == 1
+    check_task_scheduled_decision(decisions[0], increment_retry)
+
+
+def test_activity_task_timeout_raises():
+    workflow = TestDefinitionActivityRaisesOnFailure
+    executor = Executor(DOMAIN, workflow)
+
+    history = builder.History(workflow)
+    decision_id = history.last_id
+    (history
+        .add_activity_task(
+            raise_on_failure,
+            activity_id='activity-tests.test_dataflow.raise_on_failure-1',
+            decision_id=decision_id,
+            last_state='timed_out',
+            timeout_type='START_TO_CLOSE'))
+
+    decisions, _ = executor.replay(history)
+    workflow_failed = swf.models.decision.WorkflowExecutionDecision()
+    workflow_failed.fail(
+        reason='Workflow execution error in task '
+               'activity-tests.test_dataflow.raise_on_failure: '
+               '"TimeoutError(START_TO_CLOSE)"')
+
+    assert decisions[0] == workflow_failed