blob: 7c2b9f43fe4f7bed850e61dc4bb8d26ea8f5f708 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
|
import unittest
from doctest import DocTestSuite
from test import test_support
import threading
import weakref
class Weak(object):
pass
def target(local, weaklist):
weak = Weak()
local.weak = weak
weaklist.append(weakref.ref(weak))
class ThreadingLocalTest(unittest.TestCase):
def test_local_refs(self):
local = threading.local()
weaklist = []
n = 20
for i in range(n):
t = threading.Thread(target=target, args=(local, weaklist))
t.start()
t.join()
self.assertEqual(len(weaklist), n)
deadlist = [weak for weak in weaklist if weak() is None]
# XXX threading.local keeps the local of the last stopped thread alive
self.assertEqual(len(deadlist), n-1)
def test_main():
suite = unittest.TestSuite()
suite.addTest(DocTestSuite('_threading_local'))
suite.addTest(unittest.makeSuite(ThreadingLocalTest))
try:
from thread import _local
except ImportError:
pass
else:
import _threading_local
local_orig = _threading_local.local
def setUp(test):
_threading_local.local = _local
def tearDown(test):
_threading_local.local = local_orig
suite.addTest(DocTestSuite('_threading_local',
setUp=setUp, tearDown=tearDown)
)
test_support.run_unittest(suite)
if __name__ == '__main__':
test_main()
|
