Mirror
/
cpython
mirror of https://github.com/python/cpython
4
1
Fork 0
Code Issues Releases Wiki Activity

gh-90473: Skip test_queue when threading is not available (GH-93712)

This commit is contained in:
Christian Heimes 2022-06-11 11:55:11 +02:00 committed by GitHub
parent dc5e02b2f4
commit f0b7aa71c4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 2 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
Lib/test/test_queue.py
View File

@ -10,6 +10,8 @@ from test.support import gc_collect
from test.support import import_helper from test.support import import_helper
from test.support import threading_helper from test.support import threading_helper
# queue module depends on threading primitives
threading_helper.requires_working_threading(module=True)
py_queue = import_helper.import_fresh_module('queue', blocked=['_queue']) py_queue = import_helper.import_fresh_module('queue', blocked=['_queue'])
c_queue = import_helper.import_fresh_module('queue', fresh=['_queue']) c_queue = import_helper.import_fresh_module('queue', fresh=['_queue'])
@ -87,7 +89,6 @@ class BlockingTestMixin:
self.fail("trigger thread ended but event never set") self.fail("trigger thread ended but event never set")
@threading_helper.requires_working_threading()
class BaseQueueTestMixin(BlockingTestMixin): class BaseQueueTestMixin(BlockingTestMixin):
def setUp(self): def setUp(self):
self.cum = 0 self.cum = 0
@ -291,7 +292,6 @@ class CPriorityQueueTest(PriorityQueueTest, unittest.TestCase):
class FailingQueueException(Exception): pass class FailingQueueException(Exception): pass
@threading_helper.requires_working_threading()
class FailingQueueTest(BlockingTestMixin): class FailingQueueTest(BlockingTestMixin):
def setUp(self): def setUp(self):
@ -467,7 +467,6 @@ class BaseSimpleQueueTest:
return return
results.append(val) results.append(val)
@threading_helper.requires_working_threading()
def run_threads(self, n_threads, q, inputs, feed_func, consume_func): def run_threads(self, n_threads, q, inputs, feed_func, consume_func):
results = [] results = []
sentinel = None sentinel = None