1
'''Enhanced threading.Thread which can deliver a return value and propagate
2
exceptions from the called thread to the calling thread.
4
Copyright (C) 2007 Canonical Ltd.
5
Author: Martin Pitt <martin.pitt@ubuntu.com>
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.
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.'''
20
def __init__(self, group=None, target=None, name=None, args=(), kwargs={},
22
'''Initialize Thread, identical to threading.Thread.__init__().'''
24
threading.Thread.__init__(self, group, target, name, args, kwargs,
26
self.__target = target
28
self.__kwargs = kwargs
30
self._exception = None
33
'''Run target function, identical to threading.Thread.run().'''
37
self._retval = self.__target(*self.__args, **self.__kwargs)
39
self._exception = sys.exc_info()
41
def return_value(self):
42
'''Return value from target function.
44
This can only be called after the thread has finished, i. e. when
45
isAlive() is False and did not terminate with an exception.'''
47
assert not self.isAlive()
48
assert not self._exception
52
'''Return a tuple (type, value, traceback) of the exception caught in
55
return self._exception
58
'''Raises the exception caught in the thread.
60
Does nothing if no exception was caught.'''
63
raise self._exception[0], self._exception[1], self._exception[2]
69
if __name__ == '__main__':
70
import unittest, time, traceback, exceptions
73
'''Test thread to just wait a bit.'''
78
'''Test thread to divide two numbers.'''
82
class _REThreadTest(unittest.TestCase):
83
def test_return_value(self):
84
'''Test that return value works properly.'''
86
t = REThread(target=div, args=(42, 2))
89
# exc_raise() should be a no-op on successful functions
91
self.assertEqual(t.return_value(), 21)
92
self.assertEqual(t.exc_info(), None)
94
def test_no_return_value(self):
95
'''Test that REThread works if run() does not return anything.'''
97
t = REThread(target=idle, args=(0.5,))
99
# thread must be joined first
100
self.assertRaises(AssertionError, t.return_value)
102
self.assertEqual(t.return_value(), None)
103
self.assertEqual(t.exc_info(), None)
105
def test_exception(self):
106
'''Test that exception in thread is caught and passed.'''
108
t = REThread(target=div, args=(1, 0))
111
# thread did not terminate normally, no return value
112
self.assertRaises(AssertionError, t.return_value)
113
self.assert_(t.exc_info()[0] == exceptions.ZeroDivisionError)
114
exc = traceback.format_exception(t.exc_info()[0], t.exc_info()[1],
116
self.assert_(exc[-1].startswith('ZeroDivisionError'))
117
self.assert_(exc[-2].endswith('return x / y\n'))
119
def test_exc_raise(self):
120
'''Test that exc_raise() raises caught thread exception.'''
122
t = REThread(target=div, args=(1, 0))
125
# thread did not terminate normally, no return value
126
self.assertRaises(AssertionError, t.return_value)
133
exc = traceback.format_exception(e[0], e[1], e[2])
134
self.assert_(exc[-1].startswith('ZeroDivisionError'))
135
self.assert_(exc[-2].endswith('return x / y\n'))
added
removed
checkbox/contrib/REThread.py
