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