1 import unittest

     2 from doctest import DocTestSuite

     3 from test import test_support

     4 import threading

     5 import weakref

     6 import gc

     7 

     8 class Weak(object):

     9     pass

    10 

    11 def target(local, weaklist):

    12     weak = Weak()

    13     local.weak = weak

    14     weaklist.append(weakref.ref(weak))

    15 

    16 class ThreadingLocalTest(unittest.TestCase):

    17 

    18     def test_local_refs(self):

    19         self._local_refs(20)

    20         self._local_refs(50)

    21         self._local_refs(100)

    22 

    23     def _local_refs(self, n):

    24         local = threading.local()

    25         weaklist = []

    26         for i in range(n):

    27             t = threading.Thread(target=target, args=(local, weaklist))

    28             t.start()

    29             t.join()

    30         del t

    31 

    32         gc.collect()

    33         self.assertEqual(len(weaklist), n)

    34 

    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)

    38 

    39         # Assignment to the same thread local frees it sometimes (!)

    40         local.someothervar = None

    41         gc.collect()

    42         deadlist = [weak for weak in weaklist if weak() is None]

    43         self.assert_(len(deadlist) in (n-1, n), (n, len(deadlist)))

    44 

    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.

    50         import time

    51         class Local(threading.local):

    52             def __init__(self):

    53                 time.sleep(0.01)

    54         local = Local()

    55 

    56         def f(i):

    57             local.x = i

    58             # Simply check that the variable is correctly set

    59             self.assertEqual(local.x, i)

    60 

    61         threads= []

    62         for i in range(10):

    63             t = threading.Thread(target=f, args=(i,))

    64             t.start()

    65             threads.append(t)

    66 

    67         for t in threads:

    68             t.join()

    69 

    70 

    71 def test_main():

    72     suite = unittest.TestSuite()

    73     suite.addTest(DocTestSuite('_threading_local'))

    74     suite.addTest(unittest.makeSuite(ThreadingLocalTest))

    75 

    76     try:

    77         from thread import _local

    78     except ImportError:

    79         pass

    80     else:

    81         import _threading_local

    82         local_orig = _threading_local.local

    83         def setUp(test):

    84             _threading_local.local = _local

    85         def tearDown(test):

    86             _threading_local.local = local_orig

    87         suite.addTest(DocTestSuite('_threading_local',

    88                                    setUp=setUp, tearDown=tearDown)

    89                       )

    90 

    91     test_support.run_unittest(suite)

    92 

    93 if __name__ == '__main__':

    94     test_main()