mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-10-17 12:18:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 9

Issue #22042: signal.set_wakeup_fd(fd) now raises an exception if the file

Browse source 
descriptor is in blocking mode.
This commit is contained in:
Victor Stinner 2014-08-27 12:59:44 +02:00
parent 86cc17e69e
commit 3822760f2d
3 changed files with 47 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

25
Lib/test/test_signal.py
View file

@ -271,6 +271,9 @@ class WakeupFDTests(unittest.TestCase):
self.addCleanup(os.close, r2)
self.addCleanup(os.close, w2)
os.set_blocking(w1, False)
os.set_blocking(w2, False)
signal.set_wakeup_fd(w1)
self.assertEqual(signal.set_wakeup_fd(w2), w1)
self.assertEqual(signal.set_wakeup_fd(-1), w2)
@ -279,10 +282,12 @@ class WakeupFDTests(unittest.TestCase):
def test_set_wakeup_fd_socket_result(self):
sock1 = socket.socket()
self.addCleanup(sock1.close)
sock1.setblocking(False)
fd1 = sock1.fileno()
sock2 = socket.socket()
self.addCleanup(sock2.close)
sock2.setblocking(False)
fd2 = sock2.fileno()
signal.set_wakeup_fd(fd1)
@ -290,6 +295,26 @@ class WakeupFDTests(unittest.TestCase):
self.assertEqual(signal.set_wakeup_fd(-1), fd2)
self.assertEqual(signal.set_wakeup_fd(-1), -1)
# On Windows, files are always blocking and Windows does not provide a
# function to test if a socket is in non-blocking mode.
@unittest.skipIf(sys.platform == "win32", "tests specific to POSIX")
def test_set_wakeup_fd_blocking(self):
rfd, wfd = os.pipe()
self.addCleanup(os.close, rfd)
self.addCleanup(os.close, wfd)
# fd must be non-blocking
os.set_blocking(wfd, True)
with self.assertRaises(ValueError) as cm:
signal.set_wakeup_fd(wfd)
self.assertEqual(str(cm.exception),
"the fd %s must be in non-blocking mode" % wfd)
# non-blocking is ok
os.set_blocking(wfd, False)
signal.set_wakeup_fd(wfd)
signal.set_wakeup_fd(-1)
@unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
class WakeupSignalTests(unittest.TestCase):