Update for threading.local test.

This commit is contained in:
Christian Heimes 2008-01-19 15:06:09 +00:00
parent 71a0451f67
commit 95016e71ea
1 changed files with 18 additions and 2 deletions

View File

@ -3,6 +3,7 @@ from doctest import DocTestSuite
from test import test_support
import threading
import weakref
import gc
class Weak(object):
pass
@ -13,19 +14,34 @@ def target(local, weaklist):
weaklist.append(weakref.ref(weak))
class ThreadingLocalTest(unittest.TestCase):
def test_local_refs(self):
self._local_refs(20)
self._local_refs(50)
self._local_refs(100)
def _local_refs(self, n):
local = threading.local()
weaklist = []
n = 20
for i in range(n):
t = threading.Thread(target=target, args=(local, weaklist))
t.start()
t.join()
del t
gc.collect()
self.assertEqual(len(weaklist), n)
# XXX threading.local keeps the local of the last stopped thread alive.
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)
# Assignment to the same thread local frees it sometimes (!)
local.someothervar = None
gc.collect()
deadlist = [weak for weak in weaklist if weak() is None]
self.assert_(len(deadlist) in (n-1, n), (n, len(deadlist)))
def test_main():
suite = unittest.TestSuite()
suite.addTest(DocTestSuite('_threading_local'))