2 from doctest
import DocTestSuite
3 from test
import test_support
4 threading
= test_support
.import_module('threading')
11 def target(local
, weaklist
):
14 weaklist
.append(weakref
.ref(weak
))
16 class ThreadingLocalTest(unittest
.TestCase
):
18 def test_local_refs(self
):
23 def _local_refs(self
, n
):
24 local
= threading
.local()
27 t
= threading
.Thread(target
=target
, args
=(local
, weaklist
))
33 self
.assertEqual(len(weaklist
), n
)
35 # XXX threading.local keeps the local of the last stopped thread alive.
36 deadlist
= [weak
for weak
in weaklist
if weak() is None]
37 self
.assertEqual(len(deadlist
), n
-1)
39 # Assignment to the same thread local frees it sometimes (!)
40 local
.someothervar
= None
42 deadlist
= [weak
for weak
in weaklist
if weak() is None]
43 self
.assertIn(len(deadlist
), (n
-1, n
), (n
, len(deadlist
)))
45 def test_derived(self
):
46 # Issue 3088: if there is a threads switch inside the __init__
47 # of a threading.local derived class, the per-thread dictionary
48 # is created but not correctly set on the object.
49 # The first member set may be bogus.
51 class Local(threading
.local
):
58 # Simply check that the variable is correctly set
59 self
.assertEqual(local
.x
, i
)
63 t
= threading
.Thread(target
=f
, args
=(i
,))
70 def test_derived_cycle_dealloc(self
):
71 # http://bugs.python.org/issue6990
72 class Local(threading
.local
):
76 e1
= threading
.Event()
77 e2
= threading
.Event()
80 # 1) Involve Local in a cycle
85 # 2) GC the cycle (triggers threadmodule.c::local_clear
86 # before local_dealloc)
92 # 4) New Locals should be empty
93 passed
[0] = all(not hasattr(local
, 'foo') for local
in locals)
95 t
= threading
.Thread(target
=f
)
99 # 3) New Locals should recycle the original's address. Creating
100 # them in the thread overwrites the thread state and avoids the
102 locals = [Local() for i
in range(10)]
106 self
.assertTrue(passed
[0])
108 def test_arguments(self
):
110 from thread
import _local
as local
111 from _threading_local
import local
as py_local
113 for cls
in (local
, py_local
):
115 def __init__(self
, *args
, **kwargs
):
120 self
.assertRaises(TypeError, cls
, a
=1)
121 self
.assertRaises(TypeError, cls
, 1)
125 suite
= unittest
.TestSuite()
126 suite
.addTest(DocTestSuite('_threading_local'))
127 suite
.addTest(unittest
.makeSuite(ThreadingLocalTest
))
130 from thread
import _local
134 import _threading_local
135 local_orig
= _threading_local
.local
137 _threading_local
.local
= _local
139 _threading_local
.local
= local_orig
140 suite
.addTest(DocTestSuite('_threading_local',
141 setUp
=setUp
, tearDown
=tearDown
)
144 test_support
.run_unittest(suite
)
146 if __name__
== '__main__':