mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-04 08:59:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 11

bpo-40280: Detect missing threading on WASM platforms (GH-32352)

Browse source 
Co-authored-by: Brett Cannon <brett@python.org>
This commit is contained in:
Christian Heimes 2022-04-07 10:22:47 +03:00 • committed by GitHub
parent 5aee46b31b
commit 2b16a08bc7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
33 changed files with 103 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

6
Lib/test/test_io.py
View file

@ -1451,6 +1451,7 @@ class BufferedReaderTest(unittest.TestCase, CommonBufferedTests):
self.assertEqual(b"abcdefg", bufio.read())
@support.requires_resource('cpu')
@threading_helper.requires_working_threading()
def test_threads(self):
try:
# Write out many bytes with exactly the same number of 0's,
@ -1825,6 +1826,7 @@ class BufferedWriterTest(unittest.TestCase, CommonBufferedTests):
self.assertEqual(f.tell(), buffer_size + 2)
@support.requires_resource('cpu')
@threading_helper.requires_working_threading()
def test_threads(self):
try:
# Write out many bytes from many threads and test they were
@ -1895,6 +1897,7 @@ class BufferedWriterTest(unittest.TestCase, CommonBufferedTests):
self.assertRaises(OSError, b.close) # exception not swallowed
self.assertTrue(b.closed)
@threading_helper.requires_working_threading()
def test_slow_close_from_thread(self):
# Issue #31976
rawio = self.SlowFlushRawIO()
@ -3287,6 +3290,7 @@ class TextIOWrapperTest(unittest.TestCase):
self.assertEqual(f.errors, "replace")
@support.no_tracing
@threading_helper.requires_working_threading()
def test_threads_write(self):
# Issue6750: concurrent writes could duplicate data
event = threading.Event()
@ -4362,9 +4366,11 @@ class CMiscIOTest(MiscIOTest):
else:
self.assertFalse(err.strip('.!'))
@threading_helper.requires_working_threading()
def test_daemon_threads_shutdown_stdout_deadlock(self):
self.check_daemon_threads_shutdown_deadlock('stdout')
@threading_helper.requires_working_threading()
def test_daemon_threads_shutdown_stderr_deadlock(self):
self.check_daemon_threads_shutdown_deadlock('stderr')