mirror of
https://github.com/python/cpython.git
synced 2024-12-15 04:44:47 +08:00
#17249: convert a test in test_capi to use unittest and reap threads.
This commit is contained in:
parent
739d5497c0
commit
29267c81ce
@ -8,6 +8,7 @@ import random
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import _thread
|
||||
import unittest
|
||||
from test import support
|
||||
try:
|
||||
@ -222,8 +223,34 @@ class EmbeddingTest(unittest.TestCase):
|
||||
os.chdir(oldcwd)
|
||||
|
||||
|
||||
@unittest.skipUnless(threading, 'Threading required for this test.')
|
||||
class TestThreadState(unittest.TestCase):
|
||||
|
||||
@support.reap_threads
|
||||
def test_thread_state(self):
|
||||
# some extra thread-state tests driven via _testcapi
|
||||
def target():
|
||||
idents = []
|
||||
|
||||
def callback():
|
||||
idents.append(_thread.get_ident())
|
||||
|
||||
_testcapi._test_thread_state(callback)
|
||||
a = b = callback
|
||||
time.sleep(1)
|
||||
# Check our main thread is in the list exactly 3 times.
|
||||
self.assertEqual(idents.count(_thread.get_ident()), 3,
|
||||
"Couldn't find main thread correctly in the list")
|
||||
|
||||
target()
|
||||
t = threading.Thread(target=target)
|
||||
t.start()
|
||||
t.join()
|
||||
|
||||
|
||||
def test_main():
|
||||
support.run_unittest(CAPITest, TestPendingCalls, Test6012, EmbeddingTest)
|
||||
support.run_unittest(CAPITest, TestPendingCalls, Test6012,
|
||||
EmbeddingTest, TestThreadState)
|
||||
|
||||
for name in dir(_testcapi):
|
||||
if name.startswith('test_'):
|
||||
@ -232,32 +259,5 @@ def test_main():
|
||||
print("internal", name)
|
||||
test()
|
||||
|
||||
# some extra thread-state tests driven via _testcapi
|
||||
def TestThreadState():
|
||||
if support.verbose:
|
||||
print("auto-thread-state")
|
||||
|
||||
idents = []
|
||||
|
||||
def callback():
|
||||
idents.append(_thread.get_ident())
|
||||
|
||||
_testcapi._test_thread_state(callback)
|
||||
a = b = callback
|
||||
time.sleep(1)
|
||||
# Check our main thread is in the list exactly 3 times.
|
||||
if idents.count(_thread.get_ident()) != 3:
|
||||
raise support.TestFailed(
|
||||
"Couldn't find main thread correctly in the list")
|
||||
|
||||
if threading:
|
||||
import _thread
|
||||
import time
|
||||
TestThreadState()
|
||||
t = threading.Thread(target=TestThreadState)
|
||||
t.start()
|
||||
t.join()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_main()
|
||||
|
Loading…
Reference in New Issue
Block a user