More elegant test utilities for examining crosstown tasks in a django view.

Author: jMyles

.gitignore

@@ -14,3 +14,4 @@ _trial_temp*
 ignore*
 *.log
 *.log.*
+.idea

hendrix/experience/crosstown_traffic.py

@@ -3,12 +3,11 @@
 
 class ModuleCaller(object):
 
+    decorator = None
+
     def __init__(self, decorator=None):
-        if not decorator:
-            from hendrix.mechanics.async.decorators import ThroughToYou
-            self.decorator = ThroughToYou
-        else:
-            self.decorator = decorator
+        from hendrix.mechanics.async.decorators import ThroughToYou
+        self.decorator = decorator or self.decorator or ThroughToYou
 
         super(ModuleCaller, self).__init__()
 

hendrix/mechanics/async/decorators.py

@@ -36,13 +36,18 @@ def __call__(self, crosstown_task=None):
             if not self.no_go:
                 self.response.crosstown_tasks.append(self)
         except ThreadHasNoResponse:
-            if self.fail_without_response:
-                raise
-            else:
-                self.log.info("thread %s has no response; running crosstown task now.  To supress this behavior, set fail_without_response == True." % threading.current_thread())
-                self.run()
+            self.responseless_fallback(crosstown_task)
         return self.run
 
+    def responseless_fallback(self, crosstown_task=None):
+        self.crosstown_task = crosstown_task or self.crosstown_task
+
+        if self.fail_without_response:
+            raise ThreadHasNoResponse("This crosstown decorator cannot proceed without a response.  To run the crosstown_task at this time, set fail_without_response = False.")
+        else:
+            self.log.info("thread %s has no response; running crosstown task now.  To supress this behavior, set fail_without_response == True." % threading.current_thread())
+            self.run()
+
     def run(self, threadpool=None):
         if self.no_go:
             return

hendrix/test/test_testing_utils.py

@@ -1,5 +1,5 @@
 from hendrix.experience import crosstown_traffic
-from hendrix.utils.test_utils import AsyncTestMixin
+from hendrix.utils.test_utils import AsyncTestMixin, crosstownTaskListDecoratorFactory
 from twisted.trial.unittest import TestCase
 
 
@@ -48,4 +48,19 @@ def some_task():
         same_task = a_mixin.next_task()
 
         # That will be the only (and thus last) task.
-        self.assertRaises(StopIteration, a_mixin.next_task)
+        self.assertRaises(StopIteration, a_mixin.next_task)
+
+
+class TestRecordTasksAsList(TestCase):
+
+    def test_record_task_to_list(self):
+        task_list = []
+        crosstown_traffic.decorator = crosstownTaskListDecoratorFactory(task_list)
+
+        self.assertEqual(len(task_list), 0)
+
+        @crosstown_traffic()
+        def add_to_task_list():
+            print "this will be in the task list"
+
+        self.assertEqual(len(task_list), 1)

hendrix/utils/test_utils/__init__.py

@@ -3,22 +3,18 @@
 from hendrix.mechanics.async.decorators import ThroughToYou
 
 
-class MockThroughToYou(ThroughToYou):
+def crosstownTaskListDecoratorFactory(list_to_populate):
 
-    log = Logger()
-    warned_strange_overuse = False
+    class TaskListThroughToYou(ThroughToYou):
 
-    def __call__(self, crosstown_task):
-        try:
-            crosstown_traffic.recorded_tasks.append(crosstown_task)
-        except AttributeError:
-            raise TypeError('This is a MockThroughToYou object; it must be used with AsyncTestMixin or another compatible ThroughToYou object.  If this is happening in production, this is a big deal.')
+        def __init__(self, *args, **kwargs):
+            self.crosstown_task_list = list_to_populate
+            super(TaskListThroughToYou, self).__init__(*args, **kwargs)
 
-        if len(crosstown_traffic.recorded_tasks) > 10 and not self.warned_strange_overuse:
-            self.log.warning("More than 10 tasks have been recorded in a pattern meant for tests.  If this app is in production, something is wrong.")
-            self.warned_strange_overuse = True
+        def responseless_fallback(self, crosstown_task):
+            self.crosstown_task_list.append(crosstown_task)
 
-        super(MockThroughToYou, self).__call__(crosstown_task)
+    return TaskListThroughToYou
 
 
 class AsyncTestMixin(object):
@@ -28,14 +24,10 @@ def setUp(self):
         return super(AsyncTestMixin, self).setUp()
 
     def sub_setUp(self):
-        crosstown_traffic.decorator = MockThroughToYou
+        self.recorded_tasks = []
+        crosstown_traffic.decorator = crosstownTaskListDecoratorFactory(self.recorded_tasks)
         crosstown_traffic()
         self.archived_tasks = []
-        self.recorded_tasks = crosstown_traffic.recorded_tasks = []
-
-    def tearDown(self):
-        crosstown_traffic.__call__ = self.old_call
-        return super(AsyncTestMixin, self).tearDown()
 
     def next_task(self):
         for task in self.recorded_tasks: