mirrors/cpython
0
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2024-12-17 05:43:48 +08:00
Code Issues Packages Projects Releases Wiki Activity

bpo-34130: Fix test_signal.test_warn_on_full_buffer() (GH-8327)

Browse Source 
On Windows, sometimes test_signal.test_warn_on_full_buffer() fails to
fill the socketpair buffer. In that case, the C signal handler
succeed to write into the socket, it doesn't log the expected send
error, and so the test fail.

On Windows, the test now uses a timeout of 50 ms to fill the
socketpair buffer to fix this race condition.

Other changes:

* Begin with large chunk size to fill the buffer to speed up the
  test.
* Add error messages to assertion errors to more easily identify
  which assertion failed.
* Don't set the read end of the socketpair as non-blocking.
This commit is contained in:
Victor Stinner 2018-07-18 18:29:54 +02:00 committed by GitHub
parent 99bb6df66a
commit 686b4b5ff2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 41 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
Lib/test/test_signal.py
View File

@ -479,26 +479,51 @@ class WakeupSocketSignalTests(unittest.TestCase):
signal.signal(signum, handler)
read, write = socket.socketpair()
read.setblocking(False)
write.setblocking(False)
# Fill the send buffer
# Fill the socketpair buffer
if sys.platform == 'win32':
# bpo-34130: On Windows, sometimes non-blocking send fails to fill
# the full socketpair buffer, so use a timeout of 50 ms instead.
write.settimeout(0.050)
else:
write.setblocking(False)
# Start with large chunk size to reduce the
# number of send needed to fill the buffer.
written = 0
for chunk_size in (2 ** 16, 2 ** 8, 1):
chunk = b"x" * chunk_size
try:
while True:
write.send(chunk)
written += chunk_size
except (BlockingIOError, socket.timeout):
pass
print(f"%s bytes written into the socketpair" % written, flush=True)
write.setblocking(False)
try:
while True:
write.send(b"x")
write.send(b"x")
except BlockingIOError:
# The socketpair buffer seems full
pass
else:
raise AssertionError("%s bytes failed to fill the socketpair "
"buffer" % written)
# By default, we get a warning when a signal arrives
msg = ('Exception ignored when trying to {action} '
'to the signal wakeup fd')
signal.set_wakeup_fd(write.fileno())
with captured_stderr() as err:
_testcapi.raise_signal(signum)
err = err.getvalue()
if ('Exception ignored when trying to {action} to the signal wakeup fd'
not in err):
raise AssertionError(err)
if msg not in err:
raise AssertionError("first set_wakeup_fd() test failed, "
"stderr: %r" % err)
# And also if warn_on_full_buffer=True
signal.set_wakeup_fd(write.fileno(), warn_on_full_buffer=True)
@ -507,9 +532,9 @@ class WakeupSocketSignalTests(unittest.TestCase):
_testcapi.raise_signal(signum)
err = err.getvalue()
if ('Exception ignored when trying to {action} to the signal wakeup fd'
not in err):
raise AssertionError(err)
if msg not in err:
raise AssertionError("set_wakeup_fd(warn_on_full_buffer=True) "
"test failed, stderr: %r" % err)
# But not if warn_on_full_buffer=False
signal.set_wakeup_fd(write.fileno(), warn_on_full_buffer=False)
@ -519,7 +544,8 @@ class WakeupSocketSignalTests(unittest.TestCase):
err = err.getvalue()
if err != "":
raise AssertionError("got unexpected output %r" % (err,))
raise AssertionError("set_wakeup_fd(warn_on_full_buffer=False) "
"test failed, stderr: %r" % err)
# And then check the default again, to make sure warn_on_full_buffer
# settings don't leak across calls.
@ -529,9 +555,9 @@ class WakeupSocketSignalTests(unittest.TestCase):
_testcapi.raise_signal(signum)
err = err.getvalue()
if ('Exception ignored when trying to {action} to the signal wakeup fd'
not in err):
raise AssertionError(err)
if msg not in err:
raise AssertionError("second set_wakeup_fd() test failed, "
"stderr: %r" % err)
""".format(action=action)
assert_python_ok('-c', code)