home *** CD-ROM | disk | FTP | other *** search
/ Mac Easy 2010 May / Mac Life Ubuntu.iso / casper / filesystem.squashfs / usr / share / pyshared / apport / REThread.py < prev    next >
Encoding:
Python Source  |  2009-04-06  |  4.5 KB  |  141 lines
  1. '''Enhanced threading.Thread which can deliver a return value and propagate
  2. exceptions from the called thread to the calling thread.
  3.  
  4. Copyright (C) 2007 Canonical Ltd.
  5. Author: Martin Pitt <martin.pitt@ubuntu.com>
  6.  
  7. This program is free software; you can redistribute it and/or modify it
  8. under the terms of the GNU General Public License as published by the
  9. Free Software Foundation; either version 2 of the License, or (at your
  10. option) any later version.  See http://www.gnu.org/copyleft/gpl.html for
  11. the full text of the license.
  12. '''
  13.  
  14. import threading, sys
  15.  
  16. class REThread(threading.Thread):
  17.     '''Enhanced threading.Thread which can deliver a return value and propagate
  18.     exceptions from the called thread to the calling thread.'''
  19.  
  20.     def __init__(self, group=None, target=None, name=None, args=(), kwargs={},
  21.             verbose=None):
  22.         '''Initialize Thread, identical to threading.Thread.__init__().'''
  23.  
  24.         threading.Thread.__init__(self, group, target, name, args, kwargs,
  25.             verbose)
  26.         self.__target = target
  27.         self.__args = args
  28.         self.__kwargs = kwargs
  29.         self._retval = None
  30.         self._exception = None
  31.  
  32.     def run(self):
  33.         '''Run target function, identical to threading.Thread.run().'''
  34.  
  35.         if self.__target:
  36.             try:
  37.                 self._retval = self.__target(*self.__args, **self.__kwargs)
  38.             except:
  39.                 if sys:
  40.                     self._exception = sys.exc_info()
  41.  
  42.     def return_value(self):
  43.         '''Return value from target function.
  44.  
  45.         This can only be called after the thread has finished, i. e. when
  46.         isAlive() is False and did not terminate with an exception.'''
  47.  
  48.         assert not self.isAlive()
  49.         assert not self._exception
  50.         return self._retval
  51.  
  52.     def exc_info(self):
  53.         '''Return a tuple (type, value, traceback) of the exception caught in
  54.         run().'''
  55.  
  56.         return self._exception
  57.  
  58.     def exc_raise(self):
  59.         '''Raises the exception caught in the thread.
  60.  
  61.         Does nothing if no exception was caught.'''
  62.  
  63.         if self._exception:
  64.             raise self._exception[0], self._exception[1], self._exception[2]
  65.  
  66. #
  67. # Unit test
  68. #
  69.  
  70. if __name__ == '__main__':
  71.     import unittest, time, traceback, exceptions
  72.  
  73.     def idle(seconds):
  74.         '''Test thread to just wait a bit.'''
  75.  
  76.         time.sleep(seconds)
  77.  
  78.     def div(x, y):
  79.         '''Test thread to divide two numbers.'''
  80.  
  81.         return x / y
  82.  
  83.     class _REThreadTest(unittest.TestCase):
  84.         def test_return_value(self):
  85.             '''return value works properly.'''
  86.  
  87.             t = REThread(target=div, args=(42, 2))
  88.             t.start()
  89.             t.join()
  90.             # exc_raise() should be a no-op on successful functions
  91.             t.exc_raise()
  92.             self.assertEqual(t.return_value(), 21)
  93.             self.assertEqual(t.exc_info(), None)
  94.  
  95.         def test_no_return_value(self):
  96.             '''REThread works if run() does not return anything.'''
  97.  
  98.             t = REThread(target=idle, args=(0.5,))
  99.             t.start()
  100.             # thread must be joined first
  101.             self.assertRaises(AssertionError, t.return_value)
  102.             t.join()
  103.             self.assertEqual(t.return_value(), None)
  104.             self.assertEqual(t.exc_info(), None)
  105.  
  106.         def test_exception(self):
  107.             '''exception in thread is caught and passed.'''
  108.  
  109.             t = REThread(target=div, args=(1, 0))
  110.             t.start()
  111.             t.join()
  112.             # thread did not terminate normally, no return value
  113.             self.assertRaises(AssertionError, t.return_value)
  114.             self.assert_(t.exc_info()[0] == exceptions.ZeroDivisionError)
  115.             exc = traceback.format_exception(t.exc_info()[0], t.exc_info()[1],
  116.                 t.exc_info()[2])
  117.             self.assert_(exc[-1].startswith('ZeroDivisionError'))
  118.             self.assert_(exc[-2].endswith('return x / y\n'))
  119.  
  120.         def test_exc_raise(self):
  121.             '''exc_raise() raises caught thread exception.'''
  122.  
  123.             t = REThread(target=div, args=(1, 0))
  124.             t.start()
  125.             t.join()
  126.             # thread did not terminate normally, no return value
  127.             self.assertRaises(AssertionError, t.return_value)
  128.             raised = False
  129.             try:
  130.                 t.exc_raise()
  131.             except:
  132.                 raised = True
  133.                 e = sys.exc_info()
  134.                 exc = traceback.format_exception(e[0], e[1], e[2])
  135.                 self.assert_(exc[-1].startswith('ZeroDivisionError'))
  136.                 self.assert_(exc[-2].endswith('return x / y\n'))
  137.             self.assert_(raised)
  138.  
  139.     unittest.main()
  140.  
  141.