home *** CD-ROM | disk | FTP | other *** search
/ Maximum CD 2011 June / maximum-cd-2011-06.iso / DiscContents / LibO_3.3.1_Win_x86_install_multi.exe / libreoffice1.cab / test_threading_local.py < prev    next >
Encoding:
Python Source  |  2011-02-15  |  2.6 KB  |  95 lines
  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()
  95.