[ooni-probe/master] Work with aagbsn to make the TaskManager pass all tests - tor-commits

10 Mar 2013

commit 8fc8056fa1796c3f5d096d98076aa9d0bbbe47d9
Author: Arturo Filastò art@fuffa.org
Date:   Sat Jan 12 16:56:47 2013 +0100
Work with aagbsn to make the TaskManager pass all tests
Tests will retry to run until the retry limit has been reached
    Test failures are properly kept track of
---
 ooni/managers.py       |   35 +++++-----
 ooni/tasks.py          |   19 ++++--
 tests/test_managers.py |  174 ++++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 205 insertions(+), 23 deletions(-)

diff --git a/ooni/managers.py b/ooni/managers.py
index 26ac3f9..be11473 100644
--- a/ooni/managers.py
+++ b/ooni/managers.py
@@ -18,14 +18,11 @@ class TaskManager(object):
     failures = []
     concurrency = 10
+    completedTasks = 0
+
     _tasks = iter(())
     _active_tasks = []
-    def _tasksDone(self):
-        self.tasksDone.callback(None)
-
-        self.tasksDone = defer.Deferred()
-
     def _failed(self, failure, task):
         """
         The has failed to complete, we append it to the end of the task chain
@@ -34,14 +31,16 @@ class TaskManager(object):
         self._active_tasks.remove(task)
         self.failures.append((failure, task))
-        if task.failures < self.retries:
+        if task.failures <= self.retries:
             self._tasks = itertools.chain(self._tasks,
                     makeIterable(task))
-
-        self._fillSlots()
+        else:
+            task.done.callback((failure, task))
self.failed(failure, task)
+        self._fillSlots()
+
     def _fillSlots(self):
         """
         Called on test completion and schedules measurements to be run for the
@@ -52,10 +51,9 @@ class TaskManager(object):
                 task = self._tasks.next()
                 self._run(task)
             except StopIteration:
-                self._tasksDone()
                 break
-    def _suceeded(self, result, task):
+    def _succeeded(self, result, task):
         """
         We have successfully completed a measurement.
         """
@@ -64,7 +62,9 @@ class TaskManager(object):
self._fillSlots()
-        self.suceeded(result, task)
+        task.done.callback(result)
+
+        self.succeeded(result, task)
def _run(self, task):
         """
@@ -73,9 +73,9 @@ class TaskManager(object):
         """
         self._active_tasks.append(task)
-        d = task.run()
-        d.addCallback(self.succeeded, task)
-        d.addErrback(self.failed, task)
+        d = task.start()
+        d.addCallback(self._succeeded, task)
+        d.addErrback(self._failed, task)
@property
     def failedMeasurements(self):
@@ -93,13 +93,14 @@ class TaskManager(object):
         Takes as argument a single task or a task iterable and appends it to the task
         generator queue.
         """
-
         iterable = makeIterable(task_or_task_iterator)
self._tasks = itertools.chain(self._tasks, iterable)
         self._fillSlots()
def start(self):
+        self.failures = []
+
         self.tasksDone = defer.Deferred()
         self._fillSlots()
@@ -135,7 +136,7 @@ class MeasurementsManager(TaskManager):
director = None
-    def suceeded(self, result, measurement):
+    def succeeded(self, result, measurement):
         pass
def failed(self, failure, measurement):
@@ -192,7 +193,7 @@ class ReportEntryManager(object):
     def initializeTaskList(self):
         pass
-    def suceeded(self, result, measurement):
+    def succeeded(self, result, measurement):
         pass
def failed(self, failure, measurement):
diff --git a/ooni/tasks.py b/ooni/tasks.py
index 909f3e5..55f09bd 100644
--- a/ooni/tasks.py
+++ b/ooni/tasks.py
@@ -1,24 +1,31 @@
+from twisted.internet import defer
+
 class BaseTask(object):
     _timer = None
def __init__(self):
         self.running = False
         self.failures = 0
+        # This is a deferred that gets called when a test has reached it's
+        # final status, this means: all retries have been attempted or the test
+        # has successfully executed.
+        self.done = defer.Deferred()
def _failed(self, failure):
         self.failures += 1
         self.failed(failure)
-        return
+        return failure
-    def _run(self):
+    def _succeeded(self, result):
+        self.succeeded(result)
+        return result
+
+    def start(self):
         d = self.run()
         d.addErrback(self._failed)
         d.addCallback(self._succeeded)
         return d
-    def _succeeded(self, result):
-        self.succeeded(result)
-
     def succeeded(self, result):
         """
         Place here the logic to handle a successful execution of the task.
@@ -113,7 +120,7 @@ class Measurement(TaskWithTimeout):
         d.addErrback(self.failure)
         return d
-class ReportEntry(TimedOutTask):
+class ReportEntry(TaskWithTimeout):
     def __init__(self, reporter, measurement):
         self.reporter = reporter
         self.measurement = measurement
diff --git a/tests/test_managers.py b/tests/test_managers.py
new file mode 100644
index 0000000..59eba5f
--- /dev/null
+++ b/tests/test_managers.py
@@ -0,0 +1,174 @@
+from twisted.trial import unittest
+from twisted.python import failure
+from twisted.internet import defer
+
+from ooni.tasks import BaseTask
+from ooni.managers import TaskManager
+
+
+mockFailure = failure.Failure(Exception('mock'))
+
+class MockSuccessTask(BaseTask):
+    def run(self):
+        return defer.succeed(42)
+
+class MockFailTask(BaseTask):
+    def run(self):
+        return defer.fail(mockFailure)
+
+class MockFailOnceTask(BaseTask):
+    def run(self):
+        if self.failures >= 1:
+            return defer.succeed(42)
+        else:
+            return defer.fail(mockFailure)
+
+class MockTaskManager(TaskManager):
+    def __init__(self):
+        self.successes = []
+
+    def failed(self, failure, task):
+        # print "TASK"
+        # print task
+        # print "FAILURES (%s)" % task.failures
+        # print failure
+        pass
+
+    def succeeded(self, result, task):
+        self.successes.append((result, task))
+
+class TestTaskManager(unittest.TestCase):
+    def setUp(self):
+        self.taskManager = MockTaskManager()
+        self.taskManager.concurrency = 10
+        self.taskManager.retries = 2
+
+        self.taskManager.start()
+
+    def tearDown(self):
+        pass
+
+    def test_schedule_successful_one_task(self):
+        mock_task = MockSuccessTask()
+        self.taskManager.schedule(mock_task)
+
+        @mock_task.done.addCallback
+        def done(res):
+            self.assertEqual(self.taskManager.successes,
+                    [(42, mock_task)])
+        return mock_task.done
+
+    def test_schedule_failing_one_task(self):
+        mock_task = MockFailTask()
+        self.taskManager.schedule(mock_task)
+
+        @mock_task.done.addCallback
+        def done(failure):
+            self.assertEqual(len(self.taskManager.failures), 3)
+
+            self.assertEqual(failure, (mockFailure, mock_task))
+
+        return mock_task.done
+
+    def test_schedule_successful_ten_tasks(self):
+        all_done = []
+        for x in range(10):
+            mock_task = MockSuccessTask()
+            all_done.append(mock_task.done)
+            self.taskManager.schedule(mock_task)
+
+        d = defer.DeferredList(all_done)
+        @d.addCallback
+        def done(res):
+            for task_result, task_instance in self.taskManager.successes:
+                self.assertEqual(task_result, 42)
+                self.assertIsInstance(task_instance, MockSuccessTask)
+
+        return d
+
+    def test_schedule_failing_ten_tasks(self):
+        all_done = []
+        for x in range(10):
+            mock_task = MockFailTask()
+            all_done.append(mock_task.done)
+            self.taskManager.schedule(mock_task)
+
+        d = defer.DeferredList(all_done)
+        @d.addCallback
+        def done(res):
+            # 10*2 because 2 is the number of retries
+            self.assertEqual(len(self.taskManager.failures), 10*3)
+            for task_result, task_instance in self.taskManager.failures:
+                self.assertEqual(task_result, mockFailure)
+                self.assertIsInstance(task_instance, MockFailTask)
+
+        return d
+
+    def test_schedule_successful_27_tasks(self):
+        all_done = []
+        for x in range(27):
+            mock_task = MockSuccessTask()
+            all_done.append(mock_task.done)
+            self.taskManager.schedule(mock_task)
+
+        d = defer.DeferredList(all_done)
+        @d.addCallback
+        def done(res):
+            for task_result, task_instance in self.taskManager.successes:
+                self.assertEqual(task_result, 42)
+                self.assertIsInstance(task_instance, MockSuccessTask)
+
+        return d
+
+    def test_schedule_failing_27_tasks(self):
+        all_done = []
+        for x in range(27):
+            mock_task = MockFailTask()
+            all_done.append(mock_task.done)
+            self.taskManager.schedule(mock_task)
+
+        d = defer.DeferredList(all_done)
+        @d.addCallback
+        def done(res):
+            # 10*2 because 2 is the number of retries
+            self.assertEqual(len(self.taskManager.failures), 27*3)
+            for task_result, task_instance in self.taskManager.failures:
+                self.assertEqual(task_result, mockFailure)
+                self.assertIsInstance(task_instance, MockFailTask)
+
+        return d
+
+
+    def test_task_retry_and_succeed(self):
+        mock_task = MockFailOnceTask()
+        self.taskManager.schedule(mock_task)
+
+        @mock_task.done.addCallback
+        def done(res):
+            self.assertEqual(len(self.taskManager.failures), 1)
+
+            self.assertEqual(self.taskManager.failures,
+                    [(mockFailure, mock_task)])
+            self.assertEqual(self.taskManager.successes,
+                    [(42, mock_task)])
+
+        return mock_task.done
+
+    def test_task_retry_and_succeed_56_tasks(self):
+        all_done = []
+        for x in range(56):
+            mock_task = MockFailOnceTask()
+            all_done.append(mock_task.done)
+            self.taskManager.schedule(mock_task)
+
+        d = defer.DeferredList(all_done)
+        @d.addCallback
+        def done(res):
+            self.assertEqual(len(self.taskManager.failures), 56)
+
+            for task_result, task_instance in self.taskManager.successes:
+                self.assertEqual(task_result, 42)
+                self.assertIsInstance(task_instance, MockFailOnceTask)
+
+        return d
+

    