mirror of
https://github.com/python/cpython.git
synced 2025-08-04 00:48:58 +00:00
threading._DummyThread.__init__(): document obscure new code.
test_threading.test_foreign_thread(): new test does a basic check that "foreign" threads can using the threading module, and that they create a _DummyThread instance in at least one use case. This isn't a very good test, since a thread created by thread.start_new_thread() isn't particularly "foreign".
This commit is contained in:
parent
84d548994e
commit
711906e0c2
2 changed files with 31 additions and 4 deletions
|
@ -4,6 +4,7 @@ import test.test_support
|
|||
from test.test_support import verbose
|
||||
import random
|
||||
import threading
|
||||
import thread
|
||||
import time
|
||||
import unittest
|
||||
|
||||
|
@ -78,11 +79,31 @@ class ThreadTests(unittest.TestCase):
|
|||
if verbose:
|
||||
print 'waiting for all tasks to complete'
|
||||
for t in threads:
|
||||
t.join()
|
||||
t.join(NUMTASKS)
|
||||
self.assert_(not t.isAlive())
|
||||
if verbose:
|
||||
print 'all tasks done'
|
||||
self.assertEqual(numrunning.get(), 0)
|
||||
|
||||
def test_foreign_thread(self):
|
||||
# Check that a "foreign" thread can use the threading module.
|
||||
def f(mutex):
|
||||
# Acquiring an RLock forces an entry for the foreign
|
||||
# thread to get made in the threading._active map.
|
||||
r = threading.RLock()
|
||||
r.acquire()
|
||||
r.release()
|
||||
mutex.release()
|
||||
|
||||
mutex = threading.Lock()
|
||||
mutex.acquire()
|
||||
tid = thread.start_new_thread(f, (mutex,))
|
||||
# Wait for the thread to finish.
|
||||
mutex.acquire()
|
||||
self.assert_(tid in threading._active)
|
||||
self.assert_(isinstance(threading._active[tid],
|
||||
threading._DummyThread))
|
||||
del threading._active[tid]
|
||||
|
||||
def test_main():
|
||||
test.test_support.run_unittest(ThreadTests)
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue