Temporarilly add taskthread to make it easier to package for debian.

4 files changed, 637 insertions, 0 deletions

diff --git a/client/src/taskthread/__init__.py b/client/src/taskthread/__init__.py
new file mode 100755
index 00000000..a734a829
--- /dev/null
+++ b/client/src/taskthread/__init__.py
@@ -0,0 +1,296 @@
+# Copyright 2013 Hewlett-Packard Development Company, L.P.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+import logging
+import threading
+
+__version__ = '1.4'
+
+
+logger = logging.getLogger(__name__)
+
+
+class TaskInProcessException(BaseException):
+    pass
+
+
+class TaskThread(threading.Thread):
+    """
+    A thread object that repeats a task.
+
+    Usage example::
+
+        from taskthread import TaskThread
+
+        import time
+
+        def my_task(*args, **kwargs):
+            print args, kwargs
+
+        task_thread = TaskThread(my_task)
+        task_thread.start()
+        for i in xrange(10):
+            task_thread.run_task()
+            task_thread.join_task()
+        task_thread.join()
+
+    .. note:: If :py:meth:`~TaskThread.run_task` is
+        invoked while run_task is in progress,
+        :py:class:`~.TaskInProcessException` will
+        be raised.
+
+    :param task:
+        A ``function``. This param is the task to execute when
+         run_task is called.
+    :param event:
+        A ``threading.Event``. This event will be set when run_task
+         is called. The default value is a new event, but may be
+         specified for testing purposes.
+    """
+
+    daemon = True
+    '''
+    Threads marked as daemon will be terminated.
+    '''
+    def __init__(self, task, event=threading.Event(),
+                 *args, **kwargs):
+        super(TaskThread, self).__init__()
+        self.task = task
+        self.task_event = event
+        self.running = True
+        self.running_lock = threading.Lock()
+        self.in_task = False
+        self.task_complete = threading.Event()
+        self.args = args
+        self.kwargs = kwargs
+
+    def run(self):
+        """
+        Called by threading.Thread, this runs in the new thread.
+        """
+        while True:
+            self.task_event.wait()
+            if not self.running:
+                logger.debug("TaskThread exiting")
+                return
+            logger.debug("TaskThread starting task")
+            with self.running_lock:
+                self.task_event.clear()
+            self.task_complete.clear()
+            self.task(*self.args, **self.kwargs)
+            with self.running_lock:
+                self.in_task = False
+            self.task_complete.set()
+
+    def run_task(self, *args, **kwargs):
+        """
+        Run an instance of the task.
+
+        :param args:
+            The arguments to pass to the task.
+
+        :param kwargs:
+            The keyword arguments to pass to the task.
+        """
+        # Don't allow this call if the thread is currently
+        # in a task.
+        with self.running_lock:
+            if self.in_task:
+                raise TaskInProcessException()
+            self.in_task = True
+        logger.debug("Waking up the thread")
+        self.args = args
+        self.kwargs = kwargs
+        # Wake up the thread to do it's thing
+        self.task_event.set()
+
+    def join_task(self, time_out):
+        """
+        Wait for the currently running task to complete.
+
+        :param time_out:
+            An ``int``. The amount of time to wait for the
+            task to finish.
+        """
+        with self.running_lock:
+            if not self.in_task:
+                return
+
+        success = self.task_complete.wait(time_out)
+        if success:
+            self.task_complete.clear()
+        return success
+
+    def join(self, timeout=None):
+        """
+        Wait for the task to finish
+        """
+        self.running = False
+        self.task_event.set()
+        super(TaskThread, self).join(timeout=timeout)
+
+
+class TimerTask(object):
+    """
+    An object that executes a commit function at a given interval.
+    This class is driven by a TaskThread. A new TaskThread will be
+    created the first time :py:meth:`.~start` is called. All
+    subsequent calls to start will reuse the same thread.
+
+    Usage example::
+
+        from taskthread import TimerTask
+        import time
+
+        count = 0
+        def get_count():
+            return count
+        def execute():
+            print "Count: %d" % count
+
+        task = TimerTask(execute,
+                         timeout=10,
+                         count_fcn=get_count,
+                         threshold=1)
+
+        task.start()
+
+        for i in xrange(100000):
+            count += 1
+            time.sleep(1)
+        task.stop()
+        count = 0
+        task.start()
+        for i in xrange(100000):
+            count += 1
+            time.sleep(1)
+        task.shutdown()
+
+    :param execute_fcn:
+        A `function`. This function will be executed on each time interval.
+
+    :param delay:
+        An `int`. The delay in **seconds** invocations of
+        `execute_fcn`. Default: `10`.
+
+    :param count_fcn:
+        A `function`. This function returns a current count. If the count
+        has not changed more the `threshold` since the last invocation of
+        `execute_fcn`, `execute_fcn` will not be called again. If not
+        specified, `execute_fcn` will be called each time the timer fires.
+        **Optional**. If count_fcn is specified, ``threshold`` is required.
+
+    :param threshold:
+        An `int`. Specifies the minimum delta in `count_fcn` that must be
+        met for `execute_fcn` to be invoked. **Optional**. Must be
+        specified in conjunction with `count_fcn`.
+
+    """
+    def __init__(self, execute_fcn, delay=10, count_fcn=None, threshold=None):
+        self.running = True
+        self.execute_fcn = execute_fcn
+        self.last_count = 0
+        self.event = threading.Event()
+        self.delay = delay
+        self.thread = None
+        self.running_lock = threading.RLock()
+        if bool(threshold) != bool(count_fcn):
+            raise ValueError("Must specify threshold "
+                             "and count_fcn, or neither")
+
+        self.count_fcn = count_fcn
+        self.threshold = threshold
+
+    def start(self):
+        """
+        Start the task. This starts a :py:class:`.~TaskThread`, and starts
+        running run_threshold_timer on the thread.
+
+        """
+        if not self.thread:
+            logger.debug('Starting up the taskthread')
+            self.thread = TaskThread(self._run_threshold_timer)
+            self.thread.start()
+
+        if self.threshold:
+            self.last_count = 0
+
+        logger.debug('Running the task')
+        self.running = True
+        self.thread.run_task()
+
+    def stop(self):
+        """
+        Stop the task, leaving the :py:class:`.~TaskThread` running
+        so start can be called again.
+
+        """
+        logger.debug('Stopping the task')
+        wait = False
+        with self.running_lock:
+            if self.running:
+                wait = True
+                self.running = False
+        if wait:
+            self.event.set()
+            self.thread.join_task(2)
+
+    def shutdown(self):
+        """
+        Close down the task thread and stop the task if it is running.
+
+        """
+        logger.debug('Shutting down the task')
+        self.stop()
+        self.thread.join(2)
+
+    def _exec_if_threshold_met(self):
+        new_count = self.count_fcn()
+        logger.debug('new_count: %d' % new_count)
+        if new_count >= self.last_count + self.threshold:
+            self.execute_fcn()
+            self.last_count = new_count
+
+    def _exec(self):
+        if self.count_fcn:
+            self._exec_if_threshold_met()
+        else:
+            self.execute_fcn()
+
+    def _wait(self):
+        self.event.wait(timeout=self.delay)
+        self.event.clear()
+        logger.debug('Task woken up')
+
+    def _exit_loop(self):
+        """
+        If self.running is false, it means the task should shut down.
+        """
+        exit_loop = False
+        with self.running_lock:
+            if not self.running:
+                exit_loop = True
+                logger.debug('Task shutting down')
+        return exit_loop
+
+    def _run_threshold_timer(self):
+        """
+        Main loop of the timer task
+
+        """
+        logger.debug('In Task')
+        while True:
+            self._wait()
+            if self._exit_loop():
+                return
+            self._exec()
diff --git a/client/src/taskthread/tests/__init__.py b/client/src/taskthread/tests/__init__.py
new file mode 100755
index 00000000..92bd912f
--- /dev/null
+++ b/client/src/taskthread/tests/__init__.py
@@ -0,0 +1,13 @@
+# Copyright 2013 Hewlett-Packard Development Company, L.P.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
diff --git a/client/src/taskthread/tests/unit/__init__.py b/client/src/taskthread/tests/unit/__init__.py
new file mode 100755
index 00000000..92bd912f
--- /dev/null
+++ b/client/src/taskthread/tests/unit/__init__.py
@@ -0,0 +1,13 @@
+# Copyright 2013 Hewlett-Packard Development Company, L.P.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
diff --git a/client/src/taskthread/tests/unit/test_taskthread.py b/client/src/taskthread/tests/unit/test_taskthread.py
new file mode 100755
index 00000000..82565922
--- /dev/null
+++ b/client/src/taskthread/tests/unit/test_taskthread.py
@@ -0,0 +1,315 @@
+# -*- coding: utf-8 -*-
+# Copyright 2013 Hewlett-Packard Development Company, L.P.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.:w
+
+
+import threading
+import unittest2 as unittest
+
+from mock import Mock, patch
+
+from taskthread import TaskThread, TaskInProcessException, TimerTask
+
+forever_event = threading.Event()
+
+
+def forever_function(*args, **kwargs):
+    forever_event.wait()
+    forever_event.clear()
+
+
+class TaskThreadTestCase(unittest.TestCase):
+    """
+    Tests for :py:class:`.TaskThread`.
+    """
+
+    def test___init__(self):
+        """
+        Test the __init__ method. It doesn't really do much.
+        """
+        task_thread = TaskThread(forever_function)
+        self.assertEqual(forever_function, task_thread.task)
+
+    def test_run_not_running(self):
+        """
+        Verifies that thread will shut down when running is false
+        """
+        event = Mock()
+        event.wait = Mock(side_effect=[True])
+        event.clear = Mock(side_effect=Exception("Should never be called"))
+        task_thread = TaskThread(forever_function,
+                                 event=event)
+        task_thread.running = False
+        task_thread.run()
+        event.wait.assert_called_once_with()
+
+    def test_run_executes_task(self):
+        event = Mock()
+        event.wait = Mock(side_effect=[True, True])
+
+        def stop_iteration(*args, **kwargs):
+            args[0].running = False
+
+        task_thread = TaskThread(stop_iteration,
+                                 event=event)
+
+        task_thread.args = [task_thread]
+        task_thread.kwargs = {'a': 2}
+        task_thread.in_task = True
+        task_thread.run()
+        self.assertEqual(False, task_thread.in_task)
+
+    def test_run_task(self):
+        event = Mock()
+        task_thread = TaskThread(forever_function,
+                                 event=event)
+        args = [1]
+        kwargs = {'a': 1}
+
+        task_thread.run_task(*args, **kwargs)
+        self.assertEqual(tuple(args), task_thread.args)
+        self.assertEqual(kwargs, task_thread.kwargs)
+        event.set.assert_called_once_with()
+
+    def test_run_task_task_in_progress(self):
+        event = Mock()
+        task_thread = TaskThread(forever_function,
+                                 event=event)
+        task_thread.in_task = True
+        self.assertRaises(TaskInProcessException, task_thread.run_task)
+
+    def test_join_task(self):
+        task_thread = TaskThread(forever_function)
+        task_thread.in_task = True
+        task_thread.task_complete = Mock()
+        task_thread.task_complete.wait = Mock(side_effect=[True])
+        success = task_thread.join_task(1)
+        self.assertTrue(success)
+
+    def test_join_task_not_running(self):
+        task_thread = TaskThread(forever_function)
+        task_thread.task_complete = Mock()
+        task_thread.wait =\
+            Mock(side_effect=Exception("Should never be called"))
+        task_thread.join_task(1)
+
+    def test_join(self):
+        task_thread = TaskThread(forever_function)
+        task_thread.start()
+        task_thread.run_task()
+        # Set the event so the task completes
+        forever_event.set()
+        task_thread.join_task(1)
+        task_thread.join(1)
+
+    def test_execute_multiple_tasks(self):
+        task_thread = TaskThread(forever_function)
+        task_thread.start()
+        task_thread.run_task()
+        # Set the event so the task completes
+        forever_event.set()
+        task_thread.join_task(1)
+        forever_event.set()
+        task_thread.join_task(1)
+        task_thread.join(1)
+
+
+def my_func():
+    pass
+
+
+class TimerTaskTestCase(unittest.TestCase):
+
+    def test___int__(self):
+
+        task = TimerTask(my_func,
+                         delay=100)
+        self.assertEqual(my_func, task.execute_fcn)
+        self.assertEqual(100, task.delay)
+        self.assertIsNone(task.count_fcn)
+        self.assertIsNone(task.threshold)
+
+    def test___int__raises(self):
+        self.assertRaises(ValueError, TimerTask.__init__,
+                          TimerTask(None),
+                          my_func(),
+                          count_fcn=Mock())
+
+        self.assertRaises(ValueError, TimerTask.__init__,
+                          TimerTask(None),
+                          my_func(),
+                          threshold=Mock())
+
+    @patch('taskthread.TaskThread')
+    def test_start(self, TaskThreadMock):
+        task = TimerTask(my_func)
+        thread = TaskThreadMock.return_value
+
+        task.start()
+        self.assertTrue(task.running)
+        self.assertEqual(thread, task.thread)
+        thread.start.assert_called_once_with()
+        thread.run_task.assert_called_once_with()
+
+    @patch('taskthread.TaskThread')
+    def test_start_restarts(self, TaskThreadMock):
+        task = TimerTask(my_func, threshold=1, count_fcn=Mock())
+        thread = TaskThreadMock.return_value
+        task.last_count = 1
+        task.thread = thread
+
+        task.start()
+        self.assertEqual(0, task.last_count)
+        self.assertEqual(0, thread.start.called)
+        thread.run_task.assert_called_once_with()
+
+    @patch('taskthread.TaskThread')
+    def test_stop(self, TaskThreadMock):
+        running_lock = Mock()
+        running_lock.__enter__ = Mock()
+        running_lock.__exit__ = Mock()
+        task = TimerTask(my_func)
+        task.thread = TaskThreadMock.return_value
+        task.running = True
+        task.event = Mock()
+        task.running_lock = running_lock
+
+        task.stop()
+
+        self.assertEqual(False, task.running)
+        self.assertEqual(1, task.event.set.called)
+        running_lock.__enter__.assert_called_once_with()
+        running_lock.__exit__.assert_called_once_with(None, None, None)
+        task.thread.join_task.assert_called_once_with(2)
+
+    @patch('taskthread.TaskThread')
+    def test_stop_not_running(self, TaskThreadMock):
+        task = TimerTask(my_func)
+        task.thread = TaskThreadMock.return_value
+        task.running = False
+        task.event = Mock()
+
+        task.stop()
+
+        self.assertEqual(False, task.running)
+        self.assertEqual(0, task.event.set.called)
+        self.assertEqual(0, task.thread.join_task.called)
+
+    @patch('taskthread.TaskThread')
+    def test_shutdown(self, TaskThreadMock):
+        task = TimerTask(my_func)
+        task.thread = TaskThreadMock.return_value
+        task.running = False
+        task.shutdown()
+        task.thread.join.assert_called_once_with(2)
+
+    def test__exec_if_threshold_met(self):
+        self.called = False
+
+        def exec_fcn():
+            self.called = True
+
+        def count_fcn():
+            return 10
+
+        task = TimerTask(exec_fcn, count_fcn=count_fcn, threshold=1)
+        task.last_count = 9
+        task._exec_if_threshold_met()
+        self.assertTrue(self.called)
+        self.assertEqual(10, task.last_count)
+
+    def test__exec_if_threshold_met_not_met(self):
+
+        def exec_fcn():
+            raise Exception("This shouldn't happen!!")
+
+        def count_fcn():
+            return 10
+
+        task = TimerTask(exec_fcn, count_fcn=count_fcn, threshold=10)
+        task.last_count = 9
+        task._exec_if_threshold_met()
+        self.assertEqual(9, task.last_count)
+
+    def test__exec(self):
+        self.called = False
+
+        def exec_fcn():
+            self.called = True
+
+        task = TimerTask(exec_fcn)
+        task._exec()
+        self.assertTrue(self.called)
+
+    def test__exec_threshold(self):
+        self.called = False
+
+        def exec_fcn():
+            self.called = True
+
+        def count_fcn():
+            return 1
+
+        task = TimerTask(exec_fcn, count_fcn=count_fcn, threshold=1)
+        task._exec()
+        self.assertTrue(self.called)
+
+    @patch('threading.Event')
+    def test__wait(self, event_mock):
+        task = TimerTask(my_func)
+        event = event_mock.return_value
+
+        task._wait()
+        event.wait.assert_called_once_with(timeout=task.delay)
+        self.assertEqual(1, event.clear.called)
+
+    @patch('threading.RLock')
+    def test__exit_loop(self, mock_rlock):
+        task = TimerTask(my_func)
+        task.running = False
+        lock = mock_rlock.return_value
+        lock.__enter__ = Mock()
+        lock.__exit__ = Mock()
+        self.assertTrue(task._exit_loop())
+        self.assertEqual(1, lock.__enter__.called)
+        lock.__exit__.assert_called_once_with(None, None, None)
+
+    @patch('threading.RLock')
+    def test__exit_loop_running(self, mock_rlock):
+        lock = mock_rlock.return_value
+        lock.__enter__ = Mock()
+        lock.__exit__ = Mock()
+        task = TimerTask(my_func)
+        task.running = True
+        self.assertFalse(task._exit_loop())
+        self.assertEqual(1, lock.__enter__.called)
+        lock.__exit__.assert_called_once_with(None, None, None)
+
+    @patch('threading.RLock')
+    @patch('threading.Event')
+    def test__run_threshold_timer(self, event_mock, rlock_mock):
+        self.task = None
+        event = event_mock.return_value
+        lock = rlock_mock.return_value
+        lock.__enter__ = Mock()
+        lock.__exit__ = Mock()
+
+        def exec_fcn():
+            self.task.running = False
+
+        self.task = TimerTask(exec_fcn)
+        self.task._run_threshold_timer()
+
+        self.assertFalse(self.task.running)
+        self.assertEqual(2, event.wait.call_count)