Lib/test/test_threading_local.py

   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()